From 349152be9b81eff40acad4f4751aa72feb5cd734 Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <3815206+peco-engineer-bot[bot]@users.noreply.github.com>
Date: Wed, 21 Oct 2026 20:27:08 +0000
Subject: [PATCH 01/15] ArrowBatchIterator turns a failed FetchResults into
 io.EOF (#507)

Signed-off-by: peco-engineer-bot[bot] <3815206+peco-engineer-bot[bot]@users.noreply.github.com>
Signed-off-by: Vu Anh Phung <vu.phung@databricks.com>
---
 .../rows/arrowbased/arrowRecordIterator.go    | 40 ++++++++++++++---
 .../arrowbased/arrowRecordIterator_test.go    | 43 +++++++++++++++++++
 .../rows/rowscanner/resultPageIterator.go     | 22 +++++++++-
 3 files changed, 98 insertions(+), 7 deletions(-)

diff --git a/internal/rows/arrowbased/arrowRecordIterator.go b/internal/rows/arrowbased/arrowRecordIterator.go
index 820ba75f..50614345 100644
--- a/internal/rows/arrowbased/arrowRecordIterator.go
+++ b/internal/rows/arrowbased/arrowRecordIterator.go
@@ -38,6 +38,14 @@ type arrowRecordIterator struct {
 	isFinished         bool
 	arrowSchemaBytes   []byte
 	arrowSchema        *arrow.Schema
+	// err holds a page-fetch error detected while advancing the result page
+	// iterator. It must be surfaced through Next() rather than being masked as
+	// a normal end-of-results (io.EOF). See issue #507.
+	err error
+	// errReported is set once the page-fetch error has been surfaced through
+	// Next() so that checkFinished does not re-capture it from the result page
+	// iterator (whose Err() keeps returning the stored error).
+	errReported bool
 }
 
 var _ rows.ArrowBatchIterator = (*arrowRecordIterator)(nil)
@@ -49,6 +57,17 @@ func (ri *arrowRecordIterator) Next() (arrow.Record, error) {
 		return nil, io.EOF
 	}
 
+	// If advancing the result page failed, surface that error rather than
+	// masking a failed page fetch as a normal end-of-results. See #507. The
+	// error is reported once and then the iterator is finished.
+	if ri.err != nil {
+		err := ri.err
+		ri.err = nil
+		ri.errReported = true
+		ri.Close()
+		return nil, err
+	}
+
 	// make sure we have the current batch
 	err := ri.getCurrentBatch()
 	if err != nil {
@@ -66,7 +85,9 @@ func (ri *arrowRecordIterator) Next() (arrow.Record, error) {
 // Indicate whether there are any more records available
 func (ri *arrowRecordIterator) HasNext() bool {
 	ri.checkFinished()
-	return !ri.isFinished
+	// A pending page-fetch error still needs to be reported via Next(), so the
+	// iterator is not considered exhausted until that error is consumed.
+	return !ri.isFinished || ri.err != nil
 }
 
 // Free any resources associated with this iterator
@@ -88,10 +109,19 @@ func (ri *arrowRecordIterator) Close() {
 }
 
 func (ri *arrowRecordIterator) checkFinished() {
-	finished := ri.isFinished ||
-		((ri.currentBatch == nil || !ri.currentBatch.HasNext()) &&
-			(ri.batchIterator == nil || !ri.batchIterator.HasNext()) &&
-			(ri.resultPageIterator == nil || !ri.resultPageIterator.HasNext()))
+	noLocalData := (ri.currentBatch == nil || !ri.currentBatch.HasNext()) &&
+		(ri.batchIterator == nil || !ri.batchIterator.HasNext())
+
+	noMorePages := ri.resultPageIterator == nil || !ri.resultPageIterator.HasNext()
+
+	finished := ri.isFinished || (noLocalData && noMorePages)
+
+	// If the result page iterator reported no more pages because a page fetch
+	// failed (rather than because the result set was fully consumed), hold on
+	// to that error so Next() can surface it instead of io.EOF. See #507.
+	if noLocalData && noMorePages && ri.resultPageIterator != nil && ri.err == nil && !ri.errReported {
+		ri.err = ri.resultPageIterator.Err()
+	}
 
 	if finished {
 		// Reached end of result set so Close
diff --git a/internal/rows/arrowbased/arrowRecordIterator_test.go b/internal/rows/arrowbased/arrowRecordIterator_test.go
index 13891fc9..a5112d1c 100644
--- a/internal/rows/arrowbased/arrowRecordIterator_test.go
+++ b/internal/rows/arrowbased/arrowRecordIterator_test.go
@@ -193,6 +193,49 @@ func TestArrowRecordIterator(t *testing.T) {
 	})
 }
 
+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")
+}
+
 func TestArrowRecordIteratorSchema(t *testing.T) {
 	// Test with arrowSchemaBytes available
 	t.Run("schema with initial schema bytes", func(t *testing.T) {
diff --git a/internal/rows/rowscanner/resultPageIterator.go b/internal/rows/rowscanner/resultPageIterator.go
index 22f2af25..82848abc 100644
--- a/internal/rows/rowscanner/resultPageIterator.go
+++ b/internal/rows/rowscanner/resultPageIterator.go
@@ -23,6 +23,10 @@ func errRowsUnandledFetchDirection(dir string) string {
 type ResultPageIterator interface {
 	Next() (*cli_service.TFetchResultsResp, error)
 	HasNext() bool
+	// Err returns the error, if any, that caused HasNext() to report that there
+	// are no more pages. It returns nil when the iterator finished normally
+	// (i.e. the result set was fully consumed).
+	Err() error
 	Close() error
 	Delimiter
 }
@@ -112,12 +116,26 @@ type resultPageIterator struct {
 
 var _ ResultPageIterator = (*resultPageIterator)(nil)
 
+// Err returns the error, if any, that caused HasNext() to report that there are
+// no more pages. A normal end-of-results (io.EOF) is not considered an error and
+// returns nil, so callers can distinguish a failed page fetch from a fully
+// consumed result set.
+func (rpf *resultPageIterator) Err() error {
+	if rpf == nil || rpf.err == nil || rpf.err == io.EOF {
+		return nil
+	}
+	return rpf.err
+}
+
 // Returns true if there are more pages in the result set.
 func (rpf *resultPageIterator) HasNext() bool {
 	if rpf.isFinished && rpf.nextResultPage == nil {
 		// There are no more pages to load and there isn't an already fetched
-		// page waiting to retrieved by Next()
-		rpf.err = io.EOF
+		// page waiting to retrieved by Next(). Preserve a previously stored
+		// fetch error rather than masking it as a normal end-of-results.
+		if rpf.err == nil {
+			rpf.err = io.EOF
+		}
 		return false
 	}
 

From 4d90c103fc87702df3db4dd095349b4059d4f600 Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Wed, 21 Oct 2026 20:35:53 +0000
Subject: [PATCH 02/15] ai: apply changes for #509 (1 review thread)

Addresses:
  - #4341629617 at internal/rows/arrowbased/arrowRecordIterator.go:115

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
Signed-off-by: Vu Anh Phung <vu.phung@databricks.com>
---
 internal/rows/arrowbased/arrowRecordIterator.go | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/internal/rows/arrowbased/arrowRecordIterator.go b/internal/rows/arrowbased/arrowRecordIterator.go
index 50614345..7c11c7a7 100644
--- a/internal/rows/arrowbased/arrowRecordIterator.go
+++ b/internal/rows/arrowbased/arrowRecordIterator.go
@@ -112,14 +112,18 @@ func (ri *arrowRecordIterator) checkFinished() {
 	noLocalData := (ri.currentBatch == nil || !ri.currentBatch.HasNext()) &&
 		(ri.batchIterator == nil || !ri.batchIterator.HasNext())
 
-	noMorePages := ri.resultPageIterator == nil || !ri.resultPageIterator.HasNext()
+	// resultPageIterator.HasNext() issues a real FetchResults round-trip, so it
+	// is evaluated lazily — only once all local data is drained — to avoid
+	// eagerly prefetching the next page while the current one is still in use.
+	noMoreData := noLocalData &&
+		(ri.resultPageIterator == nil || !ri.resultPageIterator.HasNext())
 
-	finished := ri.isFinished || (noLocalData && noMorePages)
+	finished := ri.isFinished || noMoreData
 
 	// If the result page iterator reported no more pages because a page fetch
 	// failed (rather than because the result set was fully consumed), hold on
 	// to that error so Next() can surface it instead of io.EOF. See #507.
-	if noLocalData && noMorePages && ri.resultPageIterator != nil && ri.err == nil && !ri.errReported {
+	if noMoreData && ri.resultPageIterator != nil && ri.err == nil && !ri.errReported {
 		ri.err = ri.resultPageIterator.Err()
 	}
 

From 7d36d3564378eb358094a70697f841fe760246a3 Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Fri, 23 Oct 2026 02:05:42 +0000
Subject: [PATCH 03/15] ai: apply changes for #509 (2 review threads)

Addresses:
  - #4355373515 at internal/rows/rowscanner/resultPageIterator.go:29
  - #4355373521 at internal/rows/arrowbased/arrowRecordIterator_test.go:236

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
Signed-off-by: Vu Anh Phung <vu.phung@databricks.com>
---
 .../rows/arrowbased/arrowIPCStreamIterator.go | 43 ++++++++++++-
 .../arrowbased/arrowRecordIterator_test.go    | 64 +++++++++++++++++++
 internal/rows/rowscanner/category_test.go     | 60 +++++++++++++++++
 3 files changed, 166 insertions(+), 1 deletion(-)

diff --git a/internal/rows/arrowbased/arrowIPCStreamIterator.go b/internal/rows/arrowbased/arrowIPCStreamIterator.go
index 2aeb0d28..5d136c94 100644
--- a/internal/rows/arrowbased/arrowIPCStreamIterator.go
+++ b/internal/rows/arrowbased/arrowIPCStreamIterator.go
@@ -30,6 +30,14 @@ type arrowIPCStreamIterator struct {
 	resultPageIterator rowscanner.ResultPageIterator
 	isFinished         bool
 	arrowSchemaBytes   []byte
+	// err holds a page-fetch error detected while advancing the result page
+	// iterator. It must be surfaced through Next() rather than being masked as
+	// a normal end-of-results (io.EOF). See issue #507.
+	err error
+	// errReported is set once the page-fetch error has been surfaced through
+	// Next() so that HasNext does not re-capture it from the result page
+	// iterator (whose Err() keeps returning the stored error).
+	errReported bool
 }
 
 var _ rows.ArrowIPCStreamIterator = (*arrowIPCStreamIterator)(nil)
@@ -40,6 +48,17 @@ func (ri *arrowIPCStreamIterator) Next() (io.Reader, error) {
 		return nil, io.EOF
 	}
 
+	// If advancing the result page failed, surface that error rather than
+	// masking a failed page fetch as a normal end-of-results. See #507. The
+	// error is reported once and then the iterator is finished.
+	if ri.err != nil {
+		err := ri.err
+		ri.err = nil
+		ri.errReported = true
+		ri.Close()
+		return nil, err
+	}
+
 	if ri.ipcStreamIterator != nil && ri.ipcStreamIterator.HasNext() {
 		return ri.ipcStreamIterator.Next()
 	}
@@ -59,6 +78,12 @@ func (ri *arrowIPCStreamIterator) Next() (io.Reader, error) {
 
 // HasNext returns true if there are more streams available
 func (ri *arrowIPCStreamIterator) HasNext() bool {
+	// A pending page-fetch error still needs to be reported via Next(), so the
+	// iterator is not considered exhausted until that error is consumed.
+	if ri.err != nil {
+		return true
+	}
+
 	if ri.isFinished {
 		return false
 	}
@@ -67,7 +92,18 @@ func (ri *arrowIPCStreamIterator) HasNext() bool {
 		return true
 	}
 
-	if ri.resultPageIterator == nil || !ri.resultPageIterator.HasNext() {
+	if ri.resultPageIterator == nil {
+		return false
+	}
+
+	if !ri.resultPageIterator.HasNext() {
+		// Distinguish a failed page fetch from a fully consumed result set.
+		if !ri.errReported {
+			if err := ri.resultPageIterator.Err(); err != nil {
+				ri.err = err
+				return true
+			}
+		}
 		return false
 	}
 
@@ -102,6 +138,11 @@ func (ri *arrowIPCStreamIterator) fetchNextData() error {
 
 	if ri.resultPageIterator == nil || !ri.resultPageIterator.HasNext() {
 		ri.isFinished = true
+		if ri.resultPageIterator != nil {
+			if err := ri.resultPageIterator.Err(); err != nil {
+				return err
+			}
+		}
 		return io.EOF
 	}
 
diff --git a/internal/rows/arrowbased/arrowRecordIterator_test.go b/internal/rows/arrowbased/arrowRecordIterator_test.go
index a5112d1c..bc9389b3 100644
--- a/internal/rows/arrowbased/arrowRecordIterator_test.go
+++ b/internal/rows/arrowbased/arrowRecordIterator_test.go
@@ -234,6 +234,70 @@ func TestArrowRecordIteratorSwallowsPageError(t *testing.T) {
 	}
 	t.Logf("fetch calls=%d rows=%d lastErr=%v", calls, rows, lastErr)
 	assert.ErrorContains(t, lastErr, "connection reset")
+
+	// Every row from the direct results and the first fetched page must be
+	// delivered before the error is surfaced.
+	expectedRows := rowscanner.CountRows(executeStatementResp.DirectResults.ResultSet.Results) +
+		rowscanner.CountRows(fetchResp1.Results)
+	assert.Equal(t, expectedRows, rows)
+	assert.Equal(t, 2, calls)
+
+	// The error is reported exactly once; afterwards the iterator is finished.
+	assert.False(t, it.HasNext())
+	rec, nextErr := it.Next()
+	assert.Nil(t, rec)
+	assert.Equal(t, io.EOF, nextErr)
+	assert.False(t, it.HasNext())
+	assert.Equal(t, 2, calls)
+}
+
+func TestArrowIPCStreamIteratorSurfacesPageError(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()
+	schemaBytes := executeStatementResp.DirectResults.ResultSetMetadata.ArrowSchema
+	ipcIt, err := NewLocalIPCStreamIterator(ctx, executeStatementResp.DirectResults.ResultSet.Results.ArrowBatches, 0, schemaBytes, &cfg)
+	assert.Nil(t, err)
+	it := NewArrowIPCStreamIterator(ctx, rpi, ipcIt, schemaBytes, cfg)
+	defer it.Close()
+
+	var lastErr error
+	streams := 0
+	for it.HasNext() {
+		_, err := it.Next()
+		if err != nil {
+			lastErr = err
+			break
+		}
+		streams++
+	}
+	t.Logf("fetch calls=%d streams=%d lastErr=%v", calls, streams, lastErr)
+	assert.ErrorContains(t, lastErr, "connection reset")
+	assert.Greater(t, streams, 0)
+
+	// The error is reported once; afterwards the iterator is exhausted.
+	assert.False(t, it.HasNext())
+	_, errAfter := it.Next()
+	assert.ErrorIs(t, errAfter, io.EOF)
 }
 
 func TestArrowRecordIteratorSchema(t *testing.T) {
diff --git a/internal/rows/rowscanner/category_test.go b/internal/rows/rowscanner/category_test.go
index d5292da3..ee42f6ff 100644
--- a/internal/rows/rowscanner/category_test.go
+++ b/internal/rows/rowscanner/category_test.go
@@ -2,6 +2,7 @@ package rowscanner
 
 import (
 	"context"
+	"io"
 	"testing"
 
 	"github.com/databricks/databricks-sql-go/internal/cli_service"
@@ -35,6 +36,65 @@ func TestResultPageFetchErrorCarriesCategory(t *testing.T) {
 	assert.Equal(t, dbsqlerrint.CategoryResultSet, dbsqlerrint.CategoryFromError(err))
 }
 
+// Err() must return nil when the result set is consumed normally (io.EOF is
+// not an error).
+func TestResultPageIteratorErrNilOnNormalFinish(t *testing.T) {
+	hasMoreRows := false
+	calls := 0
+	c := &client.TestClient{
+		FnFetchResults: func(ctx context.Context, req *cli_service.TFetchResultsReq) (*cli_service.TFetchResultsResp, error) {
+			calls++
+			return &cli_service.TFetchResultsResp{
+				HasMoreRows: &hasMoreRows,
+				Results: &cli_service.TRowSet{
+					StartRowOffset: 0,
+					ArrowBatches:   []*cli_service.TSparkArrowBatch{{RowCount: 5}},
+				},
+			}, nil
+		},
+	}
+
+	rpf := NewResultPageIterator(context.Background(), NewDelimiter(0, 0), 100, nil, false, c,
+		dbsqllog.WithContext("connId", "correlationId", ""))
+
+	assert.True(t, rpf.HasNext())
+	page, err := rpf.Next()
+	assert.Nil(t, err)
+	assert.NotNil(t, page)
+
+	assert.False(t, rpf.HasNext())
+	assert.Nil(t, rpf.Err())
+	_, err = rpf.Next()
+	assert.Equal(t, io.EOF, err)
+	assert.Nil(t, rpf.Err())
+	assert.Equal(t, 1, calls)
+}
+
+// After a failed page fetch, subsequent HasNext() calls must keep the stored
+// error rather than overwriting it with io.EOF.
+func TestResultPageIteratorPreservesFetchError(t *testing.T) {
+	calls := 0
+	c := &client.TestClient{
+		FnFetchResults: func(ctx context.Context, req *cli_service.TFetchResultsReq) (*cli_service.TFetchResultsResp, error) {
+			calls++
+			return nil, errors.New("connection reset")
+		},
+	}
+
+	rpf := NewResultPageIterator(context.Background(), NewDelimiter(0, 0), 100, nil, false, c,
+		dbsqllog.WithContext("connId", "correlationId", ""))
+
+	assert.False(t, rpf.HasNext())
+	assert.ErrorContains(t, rpf.Err(), "connection reset")
+
+	// A second HasNext() must not mask the error as a normal end-of-results.
+	assert.False(t, rpf.HasNext())
+	assert.ErrorContains(t, rpf.Err(), "connection reset")
+	_, err := rpf.Next()
+	assert.ErrorContains(t, err, "connection reset")
+	assert.Equal(t, 1, calls)
+}
+
 // A page fetch that fails when the results context is already aborted must NOT
 // be tagged result_set_error.
 func TestResultPageFetchCancelNotTaggedResultSet(t *testing.T) {

From 646bc795a28fd9725a9fcb194dd6af8e3aabbcf0 Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Fri, 23 Oct 2026 02:13:15 +0000
Subject: [PATCH 04/15] ai: apply changes for #509 (1 review thread)

Addresses:
  - #4355423077 at internal/rows/arrowbased/arrowRecordIterator.go:112

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
Signed-off-by: Vu Anh Phung <vu.phung@databricks.com>
---
 .../rows/arrowbased/arrowRecordIterator.go    |  7 ++++
 .../arrowbased/arrowRecordIterator_test.go    | 41 +++++++++++++++++++
 2 files changed, 48 insertions(+)

diff --git a/internal/rows/arrowbased/arrowRecordIterator.go b/internal/rows/arrowbased/arrowRecordIterator.go
index 7c11c7a7..f1c1e82d 100644
--- a/internal/rows/arrowbased/arrowRecordIterator.go
+++ b/internal/rows/arrowbased/arrowRecordIterator.go
@@ -109,6 +109,13 @@ func (ri *arrowRecordIterator) Close() {
 }
 
 func (ri *arrowRecordIterator) checkFinished() {
+	// Once finished (including after an explicit Close()), never touch the
+	// result page iterator again: its HasNext() would issue a FetchResults
+	// round-trip against an operation that may already be closed.
+	if ri.isFinished {
+		return
+	}
+
 	noLocalData := (ri.currentBatch == nil || !ri.currentBatch.HasNext()) &&
 		(ri.batchIterator == nil || !ri.batchIterator.HasNext())
 
diff --git a/internal/rows/arrowbased/arrowRecordIterator_test.go b/internal/rows/arrowbased/arrowRecordIterator_test.go
index bc9389b3..793f6838 100644
--- a/internal/rows/arrowbased/arrowRecordIterator_test.go
+++ b/internal/rows/arrowbased/arrowRecordIterator_test.go
@@ -251,6 +251,47 @@ func TestArrowRecordIteratorSwallowsPageError(t *testing.T) {
 	assert.Equal(t, 2, calls)
 }
 
+func TestArrowRecordIteratorNoFetchAfterEarlyClose(t *testing.T) {
+	logger := dbsqllog.WithContext("connectionId", "correlationId", "")
+
+	executeStatementResp := cli_service.TExecuteStatementResp{}
+	loadTestData2(t, "directResultsMultipleFetch/ExecuteStatement.json", &executeStatementResp)
+
+	fetchCalls := 0
+	closeCalls := 0
+	tc := &client.TestClient{
+		FnFetchResults: func(ctx context.Context, req *cli_service.TFetchResultsReq) (*cli_service.TFetchResultsResp, error) {
+			fetchCalls++
+			return nil, fmt.Errorf("FetchResults: operation closed")
+		},
+		FnCloseOperation: func(ctx context.Context, req *cli_service.TCloseOperationReq) (*cli_service.TCloseOperationResp, error) {
+			closeCalls++
+			return &cli_service.TCloseOperationResp{}, nil
+		},
+	}
+
+	ctx := driverctx.NewContextWithConnId(context.Background(), "connectionId")
+	ctx = driverctx.NewContextWithCorrelationId(ctx, "correlationId")
+	rpi := rowscanner.NewResultPageIterator(ctx, rowscanner.NewDelimiter(0, 7311), 5000, nil, false, tc, logger)
+	cfg := *config.WithDefaults()
+	schemaBytes := executeStatementResp.DirectResults.ResultSetMetadata.ArrowSchema
+	bi, err := NewLocalBatchIterator(ctx, executeStatementResp.DirectResults.ResultSet.Results.ArrowBatches, 0, schemaBytes, &cfg)
+	assert.Nil(t, err)
+	it := NewArrowRecordIterator(ctx, rpi, bi, schemaBytes, cfg)
+
+	// Close before any page beyond the direct results has been read.
+	it.Close()
+	assert.Equal(t, 1, closeCalls)
+
+	// A closed iterator must not issue FetchResults or report a spurious error.
+	assert.False(t, it.HasNext())
+	rec, nextErr := it.Next()
+	assert.Nil(t, rec)
+	assert.Equal(t, io.EOF, nextErr)
+	assert.False(t, it.HasNext())
+	assert.Equal(t, 0, fetchCalls)
+}
+
 func TestArrowIPCStreamIteratorSurfacesPageError(t *testing.T) {
 	logger := dbsqllog.WithContext("connectionId", "correlationId", "")
 

From 78c978335d4e01627f6ae7363ae144dd0a5ee58a Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Fri, 23 Oct 2026 02:21:11 +0000
Subject: [PATCH 05/15] ai: apply changes for #509 (1 review thread)

Addresses:
  - #4355457211 at internal/rows/arrowbased/arrowRecordIterator_test.go:335

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
Signed-off-by: Vu Anh Phung <vu.phung@databricks.com>
---
 .../arrowbased/arrowRecordIterator_test.go    | 54 +++++++++++++++++--
 1 file changed, 51 insertions(+), 3 deletions(-)

diff --git a/internal/rows/arrowbased/arrowRecordIterator_test.go b/internal/rows/arrowbased/arrowRecordIterator_test.go
index 793f6838..417f032f 100644
--- a/internal/rows/arrowbased/arrowRecordIterator_test.go
+++ b/internal/rows/arrowbased/arrowRecordIterator_test.go
@@ -333,12 +333,60 @@ func TestArrowIPCStreamIteratorSurfacesPageError(t *testing.T) {
 	}
 	t.Logf("fetch calls=%d streams=%d lastErr=%v", calls, streams, lastErr)
 	assert.ErrorContains(t, lastErr, "connection reset")
-	assert.Greater(t, streams, 0)
 
-	// The error is reported once; afterwards the iterator is exhausted.
+	// Every stream from the direct results and the first fetched page must be
+	// delivered before the error is surfaced (one stream per local arrow batch).
+	expectedStreams := len(executeStatementResp.DirectResults.ResultSet.Results.ArrowBatches) +
+		len(fetchResp1.Results.ArrowBatches)
+	assert.Equal(t, expectedStreams, streams)
+	assert.Equal(t, 2, calls)
+
+	// The error is reported once; afterwards the iterator is exhausted and
+	// does not fetch again.
 	assert.False(t, it.HasNext())
-	_, errAfter := it.Next()
+	r, errAfter := it.Next()
+	assert.Nil(t, r)
 	assert.ErrorIs(t, errAfter, io.EOF)
+	assert.False(t, it.HasNext())
+	assert.Equal(t, 2, calls)
+}
+
+func TestArrowIPCStreamIteratorNoFetchAfterEarlyClose(t *testing.T) {
+	logger := dbsqllog.WithContext("connectionId", "correlationId", "")
+
+	executeStatementResp := cli_service.TExecuteStatementResp{}
+	loadTestData2(t, "directResultsMultipleFetch/ExecuteStatement.json", &executeStatementResp)
+
+	fetchCalls := 0
+	tc := &client.TestClient{
+		FnFetchResults: func(ctx context.Context, req *cli_service.TFetchResultsReq) (*cli_service.TFetchResultsResp, error) {
+			fetchCalls++
+			return nil, fmt.Errorf("FetchResults: operation closed")
+		},
+		FnCloseOperation: func(ctx context.Context, req *cli_service.TCloseOperationReq) (*cli_service.TCloseOperationResp, error) {
+			return &cli_service.TCloseOperationResp{}, nil
+		},
+	}
+
+	ctx := driverctx.NewContextWithConnId(context.Background(), "connectionId")
+	ctx = driverctx.NewContextWithCorrelationId(ctx, "correlationId")
+	rpi := rowscanner.NewResultPageIterator(ctx, rowscanner.NewDelimiter(0, 7311), 5000, nil, false, tc, logger)
+	cfg := *config.WithDefaults()
+	schemaBytes := executeStatementResp.DirectResults.ResultSetMetadata.ArrowSchema
+	ipcIt, err := NewLocalIPCStreamIterator(ctx, executeStatementResp.DirectResults.ResultSet.Results.ArrowBatches, 0, schemaBytes, &cfg)
+	assert.Nil(t, err)
+	it := NewArrowIPCStreamIterator(ctx, rpi, ipcIt, schemaBytes, cfg)
+
+	// Close before any page beyond the direct results has been read.
+	it.Close()
+
+	// A closed iterator must not issue FetchResults or report a spurious error.
+	assert.False(t, it.HasNext())
+	r, nextErr := it.Next()
+	assert.Nil(t, r)
+	assert.Equal(t, io.EOF, nextErr)
+	assert.False(t, it.HasNext())
+	assert.Equal(t, 0, fetchCalls)
 }
 
 func TestArrowRecordIteratorSchema(t *testing.T) {

From 6be931a0b6298ac2aa7afd8d25454dc9c76d74e1 Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Fri, 23 Oct 2026 02:29:34 +0000
Subject: [PATCH 06/15] ai: apply changes for #509 (1 review thread)

Addresses:
  - #4355495859 at internal/rows/arrowbased/arrowRecordIterator.go:88

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
Signed-off-by: Vu Anh Phung <vu.phung@databricks.com>
---
 .../rows/arrowbased/arrowRecordIterator.go    | 24 +++++++++++---
 .../arrowbased/arrowRecordIterator_test.go    | 31 +++++++++++++++++++
 2 files changed, 50 insertions(+), 5 deletions(-)

diff --git a/internal/rows/arrowbased/arrowRecordIterator.go b/internal/rows/arrowbased/arrowRecordIterator.go
index f1c1e82d..b8699b4d 100644
--- a/internal/rows/arrowbased/arrowRecordIterator.go
+++ b/internal/rows/arrowbased/arrowRecordIterator.go
@@ -60,11 +60,7 @@ func (ri *arrowRecordIterator) Next() (arrow.Record, error) {
 	// If advancing the result page failed, surface that error rather than
 	// masking a failed page fetch as a normal end-of-results. See #507. The
 	// error is reported once and then the iterator is finished.
-	if ri.err != nil {
-		err := ri.err
-		ri.err = nil
-		ri.errReported = true
-		ri.Close()
+	if err := ri.takePendingErr(); err != nil {
 		return nil, err
 	}
 
@@ -82,6 +78,19 @@ func (ri *arrowRecordIterator) Next() (arrow.Record, error) {
 	return r, err
 }
 
+// takePendingErr returns the pending page-fetch error, if any, and marks it as
+// reported so it is surfaced exactly once; the iterator is then finished.
+func (ri *arrowRecordIterator) takePendingErr() error {
+	if ri.err == nil {
+		return nil
+	}
+	err := ri.err
+	ri.err = nil
+	ri.errReported = true
+	ri.Close()
+	return err
+}
+
 // Indicate whether there are any more records available
 func (ri *arrowRecordIterator) HasNext() bool {
 	ri.checkFinished()
@@ -226,6 +235,11 @@ func (ri *arrowRecordIterator) Schema() (*arrow.Schema, error) {
 	// Try to get schema bytes if not already available
 	if ri.arrowSchemaBytes == nil {
 		if ri.HasNext() {
+			// A pending page-fetch error is reported here (once) rather than
+			// via getCurrentBatch, so a later Next() does not repeat it.
+			if err := ri.takePendingErr(); err != nil {
+				return nil, err
+			}
 			if err := ri.getCurrentBatch(); err != nil {
 				return nil, err
 			}
diff --git a/internal/rows/arrowbased/arrowRecordIterator_test.go b/internal/rows/arrowbased/arrowRecordIterator_test.go
index 417f032f..f4f86ce9 100644
--- a/internal/rows/arrowbased/arrowRecordIterator_test.go
+++ b/internal/rows/arrowbased/arrowRecordIterator_test.go
@@ -251,6 +251,37 @@ func TestArrowRecordIteratorSwallowsPageError(t *testing.T) {
 	assert.Equal(t, 2, calls)
 }
 
+func TestArrowRecordIteratorSchemaReportsPageErrorOnce(t *testing.T) {
+	logger := dbsqllog.WithContext("connectionId", "correlationId", "")
+
+	calls := 0
+	failing := &client.TestClient{
+		FnFetchResults: func(ctx context.Context, req *cli_service.TFetchResultsReq) (*cli_service.TFetchResultsResp, error) {
+			calls++
+			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()
+	// nil batch iterator and nil schema bytes: Schema() must fetch a page.
+	it := NewArrowRecordIterator(ctx, rpi, nil, nil, cfg)
+	defer it.Close()
+
+	schema, err := it.Schema()
+	assert.Nil(t, schema)
+	assert.ErrorContains(t, err, "connection reset")
+
+	// The error was reported by Schema(); Next() must not report it again.
+	assert.False(t, it.HasNext())
+	rec, nextErr := it.Next()
+	assert.Nil(t, rec)
+	assert.Equal(t, io.EOF, nextErr)
+	assert.Equal(t, 1, calls)
+}
+
 func TestArrowRecordIteratorNoFetchAfterEarlyClose(t *testing.T) {
 	logger := dbsqllog.WithContext("connectionId", "correlationId", "")
 

From 93d9e7c19080f7a585c750f99d3b65e0852e09a1 Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Fri, 23 Oct 2026 02:38:33 +0000
Subject: [PATCH 07/15] ai: apply changes for #509 (1 review thread)

Addresses:
  - #4355534513 at internal/rows/arrowbased/arrowRecordIterator.go:99

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
Signed-off-by: Vu Anh Phung <vu.phung@databricks.com>
---
 .../rows/arrowbased/arrowIPCStreamIterator.go |  6 +-
 .../rows/arrowbased/arrowRecordIterator.go    | 18 ++++-
 .../arrowbased/arrowRecordIterator_test.go    | 68 +++++++++++++++++++
 3 files changed, 88 insertions(+), 4 deletions(-)

diff --git a/internal/rows/arrowbased/arrowIPCStreamIterator.go b/internal/rows/arrowbased/arrowIPCStreamIterator.go
index 5d136c94..0838c812 100644
--- a/internal/rows/arrowbased/arrowIPCStreamIterator.go
+++ b/internal/rows/arrowbased/arrowIPCStreamIterator.go
@@ -110,8 +110,12 @@ func (ri *arrowIPCStreamIterator) HasNext() bool {
 	return true
 }
 
-// Close releases resources
+// Close releases resources. It also discards any pending, not-yet-reported
+// page-fetch error so a closed iterator reports HasNext()==false and
+// Next()==io.EOF.
 func (ri *arrowIPCStreamIterator) Close() {
+	ri.err = nil
+	ri.errReported = true
 	if ri.ipcStreamIterator != nil {
 		ri.ipcStreamIterator.Close()
 		ri.ipcStreamIterator = nil
diff --git a/internal/rows/arrowbased/arrowRecordIterator.go b/internal/rows/arrowbased/arrowRecordIterator.go
index b8699b4d..f081686b 100644
--- a/internal/rows/arrowbased/arrowRecordIterator.go
+++ b/internal/rows/arrowbased/arrowRecordIterator.go
@@ -99,8 +99,19 @@ func (ri *arrowRecordIterator) HasNext() bool {
 	return !ri.isFinished || ri.err != nil
 }
 
-// Free any resources associated with this iterator
+// Free any resources associated with this iterator. An explicit Close also
+// discards any pending, not-yet-reported page-fetch error so a closed iterator
+// reports HasNext()==false and Next()==io.EOF.
 func (ri *arrowRecordIterator) Close() {
+	ri.err = nil
+	ri.errReported = true
+	ri.release()
+}
+
+// release frees the iterator's resources and marks it finished without
+// touching any pending page-fetch error, so that error can still be surfaced
+// via Next() when the end of the result set is reached internally.
+func (ri *arrowRecordIterator) release() {
 	if !ri.isFinished {
 		ri.isFinished = true
 		if ri.currentBatch != nil {
@@ -144,8 +155,9 @@ func (ri *arrowRecordIterator) checkFinished() {
 	}
 
 	if finished {
-		// Reached end of result set so Close
-		ri.Close()
+		// Reached end of result set so release resources, keeping any
+		// pending page-fetch error for Next() to report.
+		ri.release()
 	}
 }
 
diff --git a/internal/rows/arrowbased/arrowRecordIterator_test.go b/internal/rows/arrowbased/arrowRecordIterator_test.go
index f4f86ce9..16a3d8b6 100644
--- a/internal/rows/arrowbased/arrowRecordIterator_test.go
+++ b/internal/rows/arrowbased/arrowRecordIterator_test.go
@@ -323,6 +323,74 @@ func TestArrowRecordIteratorNoFetchAfterEarlyClose(t *testing.T) {
 	assert.Equal(t, 0, fetchCalls)
 }
 
+func TestArrowRecordIteratorCloseDiscardsPendingPageError(t *testing.T) {
+	logger := dbsqllog.WithContext("connectionId", "correlationId", "")
+
+	calls := 0
+	failing := &client.TestClient{
+		FnFetchResults: func(ctx context.Context, req *cli_service.TFetchResultsReq) (*cli_service.TFetchResultsResp, error) {
+			calls++
+			return nil, fmt.Errorf("FetchResults: connection reset")
+		},
+		FnCloseOperation: func(ctx context.Context, req *cli_service.TCloseOperationReq) (*cli_service.TCloseOperationResp, error) {
+			return &cli_service.TCloseOperationResp{}, nil
+		},
+	}
+
+	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()
+	it := NewArrowRecordIterator(ctx, rpi, nil, nil, cfg)
+
+	// HasNext captures the page-fetch error as pending.
+	assert.True(t, it.HasNext())
+	assert.Equal(t, 1, calls)
+
+	// An explicit Close discards the pending error.
+	it.Close()
+	assert.False(t, it.HasNext())
+	rec, nextErr := it.Next()
+	assert.Nil(t, rec)
+	assert.Equal(t, io.EOF, nextErr)
+	assert.False(t, it.HasNext())
+	assert.Equal(t, 1, calls)
+}
+
+func TestArrowIPCStreamIteratorCloseDiscardsPendingPageError(t *testing.T) {
+	logger := dbsqllog.WithContext("connectionId", "correlationId", "")
+
+	calls := 0
+	failing := &client.TestClient{
+		FnFetchResults: func(ctx context.Context, req *cli_service.TFetchResultsReq) (*cli_service.TFetchResultsResp, error) {
+			calls++
+			return nil, fmt.Errorf("FetchResults: connection reset")
+		},
+		FnCloseOperation: func(ctx context.Context, req *cli_service.TCloseOperationReq) (*cli_service.TCloseOperationResp, error) {
+			return &cli_service.TCloseOperationResp{}, nil
+		},
+	}
+
+	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()
+	it := NewArrowIPCStreamIterator(ctx, rpi, nil, nil, cfg)
+
+	// HasNext captures the page-fetch error as pending.
+	assert.True(t, it.HasNext())
+	assert.Equal(t, 1, calls)
+
+	// An explicit Close discards the pending error.
+	it.Close()
+	assert.False(t, it.HasNext())
+	r, nextErr := it.Next()
+	assert.Nil(t, r)
+	assert.Equal(t, io.EOF, nextErr)
+	assert.False(t, it.HasNext())
+	assert.Equal(t, 1, calls)
+}
+
 func TestArrowIPCStreamIteratorSurfacesPageError(t *testing.T) {
 	logger := dbsqllog.WithContext("connectionId", "correlationId", "")
 

From 085a976fd4a5eb7d648ae74cfc1885ab7d5aba88 Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Fri, 23 Oct 2026 02:45:36 +0000
Subject: [PATCH 08/15] ai: apply changes for #509 (1 review thread)

Addresses:
  - #4355577382 at internal/rows/arrowbased/arrowIPCStreamIterator.go:145

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
Signed-off-by: Vu Anh Phung <vu.phung@databricks.com>
---
 .../rows/arrowbased/arrowIPCStreamIterator.go | 28 ++++++++++++-------
 1 file changed, 18 insertions(+), 10 deletions(-)

diff --git a/internal/rows/arrowbased/arrowIPCStreamIterator.go b/internal/rows/arrowbased/arrowIPCStreamIterator.go
index 0838c812..d60144ea 100644
--- a/internal/rows/arrowbased/arrowIPCStreamIterator.go
+++ b/internal/rows/arrowbased/arrowIPCStreamIterator.go
@@ -51,11 +51,7 @@ func (ri *arrowIPCStreamIterator) Next() (io.Reader, error) {
 	// If advancing the result page failed, surface that error rather than
 	// masking a failed page fetch as a normal end-of-results. See #507. The
 	// error is reported once and then the iterator is finished.
-	if ri.err != nil {
-		err := ri.err
-		ri.err = nil
-		ri.errReported = true
-		ri.Close()
+	if err := ri.takePendingErr(); err != nil {
 		return nil, err
 	}
 
@@ -76,6 +72,20 @@ func (ri *arrowIPCStreamIterator) Next() (io.Reader, error) {
 	return nil, io.EOF
 }
 
+// takePendingErr returns the pending page-fetch error, if any, and marks it as
+// reported so it is surfaced exactly once; the iterator is then finished.
+// Mirrors arrowRecordIterator.takePendingErr.
+func (ri *arrowIPCStreamIterator) takePendingErr() error {
+	if ri.err == nil {
+		return nil
+	}
+	err := ri.err
+	ri.err = nil
+	ri.errReported = true
+	ri.Close()
+	return err
+}
+
 // HasNext returns true if there are more streams available
 func (ri *arrowIPCStreamIterator) HasNext() bool {
 	// A pending page-fetch error still needs to be reported via Next(), so the
@@ -141,12 +151,10 @@ func (ri *arrowIPCStreamIterator) fetchNextData() error {
 	}
 
 	if ri.resultPageIterator == nil || !ri.resultPageIterator.HasNext() {
+		// A page-fetch error (resultPageIterator.Err()) cannot reach here: Next()
+		// always calls HasNext() first, which captures it into ri.err and
+		// Next() surfaces it via takePendingErr() before calling fetchNextData.
 		ri.isFinished = true
-		if ri.resultPageIterator != nil {
-			if err := ri.resultPageIterator.Err(); err != nil {
-				return err
-			}
-		}
 		return io.EOF
 	}
 

From bc6582fe460ec45249bbb9d4ce0c33d39313d318 Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Mon, 26 Oct 2026 18:36:31 +0000
Subject: [PATCH 09/15] ai: apply changes for #509 (2 review threads)

Addresses:
  - #4380986647 at internal/rows/arrowbased/arrowRecordIterator.go:148
  - #4380986658 at internal/rows/arrowbased/arrowIPCStreamIterator.go:111

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
---
 .../rows/arrowbased/arrowIPCStreamIterator.go  | 18 ++++++------------
 .../rows/arrowbased/arrowRecordIterator.go     | 16 +++++-----------
 2 files changed, 11 insertions(+), 23 deletions(-)

diff --git a/internal/rows/arrowbased/arrowIPCStreamIterator.go b/internal/rows/arrowbased/arrowIPCStreamIterator.go
index d60144ea..d47bf96d 100644
--- a/internal/rows/arrowbased/arrowIPCStreamIterator.go
+++ b/internal/rows/arrowbased/arrowIPCStreamIterator.go
@@ -32,12 +32,10 @@ type arrowIPCStreamIterator struct {
 	arrowSchemaBytes   []byte
 	// err holds a page-fetch error detected while advancing the result page
 	// iterator. It must be surfaced through Next() rather than being masked as
-	// a normal end-of-results (io.EOF). See issue #507.
+	// a normal end-of-results (io.EOF). See issue #507. Once reported (or on
+	// Close) the iterator is finished, so HasNext never re-captures it from the
+	// result page iterator.
 	err error
-	// errReported is set once the page-fetch error has been surfaced through
-	// Next() so that HasNext does not re-capture it from the result page
-	// iterator (whose Err() keeps returning the stored error).
-	errReported bool
 }
 
 var _ rows.ArrowIPCStreamIterator = (*arrowIPCStreamIterator)(nil)
@@ -81,7 +79,6 @@ func (ri *arrowIPCStreamIterator) takePendingErr() error {
 	}
 	err := ri.err
 	ri.err = nil
-	ri.errReported = true
 	ri.Close()
 	return err
 }
@@ -108,11 +105,9 @@ func (ri *arrowIPCStreamIterator) HasNext() bool {
 
 	if !ri.resultPageIterator.HasNext() {
 		// Distinguish a failed page fetch from a fully consumed result set.
-		if !ri.errReported {
-			if err := ri.resultPageIterator.Err(); err != nil {
-				ri.err = err
-				return true
-			}
+		if err := ri.resultPageIterator.Err(); err != nil {
+			ri.err = err
+			return true
 		}
 		return false
 	}
@@ -125,7 +120,6 @@ func (ri *arrowIPCStreamIterator) HasNext() bool {
 // Next()==io.EOF.
 func (ri *arrowIPCStreamIterator) Close() {
 	ri.err = nil
-	ri.errReported = true
 	if ri.ipcStreamIterator != nil {
 		ri.ipcStreamIterator.Close()
 		ri.ipcStreamIterator = nil
diff --git a/internal/rows/arrowbased/arrowRecordIterator.go b/internal/rows/arrowbased/arrowRecordIterator.go
index f081686b..999bd42b 100644
--- a/internal/rows/arrowbased/arrowRecordIterator.go
+++ b/internal/rows/arrowbased/arrowRecordIterator.go
@@ -40,12 +40,10 @@ type arrowRecordIterator struct {
 	arrowSchema        *arrow.Schema
 	// err holds a page-fetch error detected while advancing the result page
 	// iterator. It must be surfaced through Next() rather than being masked as
-	// a normal end-of-results (io.EOF). See issue #507.
+	// a normal end-of-results (io.EOF). See issue #507. Once captured the
+	// iterator is finished, so checkFinished never re-captures it from the
+	// result page iterator.
 	err error
-	// errReported is set once the page-fetch error has been surfaced through
-	// Next() so that checkFinished does not re-capture it from the result page
-	// iterator (whose Err() keeps returning the stored error).
-	errReported bool
 }
 
 var _ rows.ArrowBatchIterator = (*arrowRecordIterator)(nil)
@@ -86,7 +84,6 @@ func (ri *arrowRecordIterator) takePendingErr() error {
 	}
 	err := ri.err
 	ri.err = nil
-	ri.errReported = true
 	ri.Close()
 	return err
 }
@@ -104,7 +101,6 @@ func (ri *arrowRecordIterator) HasNext() bool {
 // reports HasNext()==false and Next()==io.EOF.
 func (ri *arrowRecordIterator) Close() {
 	ri.err = nil
-	ri.errReported = true
 	ri.release()
 }
 
@@ -145,16 +141,14 @@ func (ri *arrowRecordIterator) checkFinished() {
 	noMoreData := noLocalData &&
 		(ri.resultPageIterator == nil || !ri.resultPageIterator.HasNext())
 
-	finished := ri.isFinished || noMoreData
-
 	// If the result page iterator reported no more pages because a page fetch
 	// failed (rather than because the result set was fully consumed), hold on
 	// to that error so Next() can surface it instead of io.EOF. See #507.
-	if noMoreData && ri.resultPageIterator != nil && ri.err == nil && !ri.errReported {
+	if noMoreData && ri.resultPageIterator != nil && ri.err == nil {
 		ri.err = ri.resultPageIterator.Err()
 	}
 
-	if finished {
+	if noMoreData {
 		// Reached end of result set so release resources, keeping any
 		// pending page-fetch error for Next() to report.
 		ri.release()

From b94e0004523cca6b6558ee31ba7cc5e70235cb2d Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Mon, 26 Oct 2026 18:43:29 +0000
Subject: [PATCH 10/15] ai: apply changes for #509 (1 review thread)

Addresses:
  - #4381078811 at internal/rows/rowscanner/resultPageIterator.go:29

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
---
 internal/rows/rows.go | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/internal/rows/rows.go b/internal/rows/rows.go
index 1ba3bb7c..2b62b55e 100644
--- a/internal/rows/rows.go
+++ b/internal/rows/rows.go
@@ -33,6 +33,10 @@ type rows struct {
 	// The RowScanner is responsible for handling the different
 	// formats in which the query results can be returned
 	rowscanner.RowScanner
+	// NOTE: embedding promotes ResultPageIterator.Err() onto *rows. That
+	// method reports only the page iterator's last fetch error; it is NOT a
+	// rows-level error accessor. Use iterationErr for the first error seen
+	// during row iteration.
 	rowscanner.ResultPageIterator
 
 	// Handle for the associated database operation.

From 835bdc70ad4401f7f55f74bb850c45ca379d4263 Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Mon, 26 Oct 2026 18:51:19 +0000
Subject: [PATCH 11/15] ai: apply changes for #509 (1 review thread)

Addresses:
  - #4381145698 at internal/rows/rows.go:36

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
---
 internal/rows/rows.go      | 17 +++++++++++++----
 internal/rows/rows_test.go | 16 ++++++++++++++++
 2 files changed, 29 insertions(+), 4 deletions(-)

diff --git a/internal/rows/rows.go b/internal/rows/rows.go
index 2b62b55e..467f5c8f 100644
--- a/internal/rows/rows.go
+++ b/internal/rows/rows.go
@@ -33,10 +33,8 @@ type rows struct {
 	// The RowScanner is responsible for handling the different
 	// formats in which the query results can be returned
 	rowscanner.RowScanner
-	// NOTE: embedding promotes ResultPageIterator.Err() onto *rows. That
-	// method reports only the page iterator's last fetch error; it is NOT a
-	// rows-level error accessor. Use iterationErr for the first error seen
-	// during row iteration.
+	// NOTE: ResultPageIterator.Err() is shadowed by (*rows).Err below so the
+	// page iterator's last-fetch error is not promoted onto *rows.
 	rowscanner.ResultPageIterator
 
 	// Handle for the associated database operation.
@@ -91,6 +89,17 @@ type rows struct {
 	iterationErr       error // first error from Next()/fetchResultPage, passed to closeCallback
 }
 
+// Err returns the first error seen during row iteration (from Next() or a
+// result page fetch), or nil. It deliberately shadows the embedded
+// ResultPageIterator.Err(), which reports only the page iterator's last fetch
+// error and would be misleading as a rows-level accessor.
+func (r *rows) Err() error {
+	if r == nil {
+		return nil
+	}
+	return r.iterationErr
+}
+
 var _ driver.Rows = (*rows)(nil)
 var _ driver.RowsColumnTypeScanType = (*rows)(nil)
 var _ driver.RowsColumnTypeDatabaseTypeName = (*rows)(nil)
diff --git a/internal/rows/rows_test.go b/internal/rows/rows_test.go
index e04f9ef5..b1b90024 100644
--- a/internal/rows/rows_test.go
+++ b/internal/rows/rows_test.go
@@ -1564,6 +1564,22 @@ func TestFetchResultPage_PropagatesGetNextPageError(t *testing.T) {
 	assert.ErrorContains(t, actualErr, errorMsg)
 }
 
+// (*rows).Err must shadow the embedded ResultPageIterator.Err() and report the
+// rows-level iteration error.
+func TestRowsErr_ShadowsPageIteratorErr(t *testing.T) {
+	var nilRows *rows
+	assert.NoError(t, nilRows.Err())
+
+	sentinel := errors.New("iteration failed")
+	r := &rows{iterationErr: sentinel}
+	errer, ok := interface{}(r).(interface{ Err() error })
+	assert.True(t, ok)
+	assert.Equal(t, sentinel, errer.Err())
+
+	r = &rows{}
+	assert.NoError(t, r.Err())
+}
+
 func TestNewRows_DetachesResultRPCContextFromQueryContextCancellation(t *testing.T) {
 	t.Parallel()
 

From 2198acb77605f51dd845468ab63ca0e82cb20ef1 Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Mon, 26 Oct 2026 18:58:57 +0000
Subject: [PATCH 12/15] ai: apply changes for #509 (1 review thread)

Addresses:
  - #4381227799 at internal/rows/rows.go:96

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
---
 internal/rows/rows.go      | 13 -------------
 internal/rows/rows_test.go | 16 ----------------
 2 files changed, 29 deletions(-)

diff --git a/internal/rows/rows.go b/internal/rows/rows.go
index 467f5c8f..1ba3bb7c 100644
--- a/internal/rows/rows.go
+++ b/internal/rows/rows.go
@@ -33,8 +33,6 @@ type rows struct {
 	// The RowScanner is responsible for handling the different
 	// formats in which the query results can be returned
 	rowscanner.RowScanner
-	// NOTE: ResultPageIterator.Err() is shadowed by (*rows).Err below so the
-	// page iterator's last-fetch error is not promoted onto *rows.
 	rowscanner.ResultPageIterator
 
 	// Handle for the associated database operation.
@@ -89,17 +87,6 @@ type rows struct {
 	iterationErr       error // first error from Next()/fetchResultPage, passed to closeCallback
 }
 
-// Err returns the first error seen during row iteration (from Next() or a
-// result page fetch), or nil. It deliberately shadows the embedded
-// ResultPageIterator.Err(), which reports only the page iterator's last fetch
-// error and would be misleading as a rows-level accessor.
-func (r *rows) Err() error {
-	if r == nil {
-		return nil
-	}
-	return r.iterationErr
-}
-
 var _ driver.Rows = (*rows)(nil)
 var _ driver.RowsColumnTypeScanType = (*rows)(nil)
 var _ driver.RowsColumnTypeDatabaseTypeName = (*rows)(nil)
diff --git a/internal/rows/rows_test.go b/internal/rows/rows_test.go
index b1b90024..e04f9ef5 100644
--- a/internal/rows/rows_test.go
+++ b/internal/rows/rows_test.go
@@ -1564,22 +1564,6 @@ func TestFetchResultPage_PropagatesGetNextPageError(t *testing.T) {
 	assert.ErrorContains(t, actualErr, errorMsg)
 }
 
-// (*rows).Err must shadow the embedded ResultPageIterator.Err() and report the
-// rows-level iteration error.
-func TestRowsErr_ShadowsPageIteratorErr(t *testing.T) {
-	var nilRows *rows
-	assert.NoError(t, nilRows.Err())
-
-	sentinel := errors.New("iteration failed")
-	r := &rows{iterationErr: sentinel}
-	errer, ok := interface{}(r).(interface{ Err() error })
-	assert.True(t, ok)
-	assert.Equal(t, sentinel, errer.Err())
-
-	r = &rows{}
-	assert.NoError(t, r.Err())
-}
-
 func TestNewRows_DetachesResultRPCContextFromQueryContextCancellation(t *testing.T) {
 	t.Parallel()
 

From bbacd1366f0c877f9496a78dd36df0bae845595c Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Mon, 26 Oct 2026 19:05:38 +0000
Subject: [PATCH 13/15] ai: apply changes for #509 (1 review thread)

Addresses:
  - #4381297380 at internal/rows/arrowbased/arrowIPCStreamIterator.go:148

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
---
 internal/rows/arrowbased/arrowIPCStreamIterator.go | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

diff --git a/internal/rows/arrowbased/arrowIPCStreamIterator.go b/internal/rows/arrowbased/arrowIPCStreamIterator.go
index d47bf96d..09da1e65 100644
--- a/internal/rows/arrowbased/arrowIPCStreamIterator.go
+++ b/internal/rows/arrowbased/arrowIPCStreamIterator.go
@@ -145,10 +145,16 @@ func (ri *arrowIPCStreamIterator) fetchNextData() error {
 	}
 
 	if ri.resultPageIterator == nil || !ri.resultPageIterator.HasNext() {
-		// A page-fetch error (resultPageIterator.Err()) cannot reach here: Next()
-		// always calls HasNext() first, which captures it into ri.err and
-		// Next() surfaces it via takePendingErr() before calling fetchNextData.
+		// Normally a page-fetch error is captured by HasNext() and surfaced by
+		// Next() via takePendingErr() before we get here. Check it anyway so a
+		// failed fetch can never be masked as io.EOF (#507) if call ordering
+		// changes.
 		ri.isFinished = true
+		if ri.resultPageIterator != nil {
+			if err := ri.resultPageIterator.Err(); err != nil {
+				return err
+			}
+		}
 		return io.EOF
 	}
 

From f70e409f20662445ab354420bcccb5546a05fc49 Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Mon, 26 Oct 2026 19:13:53 +0000
Subject: [PATCH 14/15] ai: apply changes for #509 (1 review thread)

Addresses:
  - #4381358155 at internal/rows/rowscanner/resultPageIterator.go:136

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
---
 internal/rows/rows_test.go | 33 +++++++++++++++++++++++++++++++++
 1 file changed, 33 insertions(+)

diff --git a/internal/rows/rows_test.go b/internal/rows/rows_test.go
index e04f9ef5..379a6c2e 100644
--- a/internal/rows/rows_test.go
+++ b/internal/rows/rows_test.go
@@ -1564,6 +1564,39 @@ func TestFetchResultPage_PropagatesGetNextPageError(t *testing.T) {
 	assert.ErrorContains(t, actualErr, errorMsg)
 }
 
+// Pins the rows-level contract after a failed page fetch: the fetch error is
+// sticky (every later Next() returns the same error rather than a clean io.EOF),
+// the failed fetch is not retried, and the error is what trackIterationErr
+// records for the close callback / telemetry.
+func TestRowsNext_FetchErrorIsStickyNotEOF(t *testing.T) {
+	errorMsg := "connection reset during FetchResults"
+	calls := 0
+	c := &client.TestClient{
+		FnFetchResults: func(ctx context.Context, req *cli_service.TFetchResultsReq) (*cli_service.TFetchResultsResp, error) {
+			calls++
+			return nil, errors.New(errorMsg)
+		},
+	}
+
+	ctx := driverctx.NewContextWithConnId(context.Background(), "connId")
+	ctx = driverctx.NewContextWithCorrelationId(ctx, "corrId")
+
+	cfg := config.WithDefaults()
+	dr, dbErr := NewRows(ctx, nil, c, cfg, nil, nil)
+	assert.Nil(t, dbErr)
+	r, ok := dr.(*rows)
+	assert.True(t, ok)
+
+	for i := 0; i < 3; i++ {
+		err := r.Next(nil)
+		assert.ErrorContains(t, err, errorMsg, "call %d", i)
+		assert.NotEqual(t, io.EOF, err, "call %d", i)
+	}
+
+	assert.Equal(t, 1, calls, "a failed page fetch must not be retried")
+	assert.ErrorContains(t, r.iterationErr, errorMsg)
+}
+
 func TestNewRows_DetachesResultRPCContextFromQueryContextCancellation(t *testing.T) {
 	t.Parallel()
 

From 62e1e964e34d03bb337bd4650f58f37a4a163b95 Mon Sep 17 00:00:00 2001
From: "peco-engineer-bot[bot]"
 <peco-engineer-bot[bot]@users.noreply.github.com>
Date: Tue, 27 Oct 2026 00:21:05 +0000
Subject: [PATCH 15/15] ai: apply changes for #509 (1 review thread)

Addresses:
  - #4383853749 at internal/rows/arrowbased/arrowRecordIterator_test.go:196

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>
---
 internal/rows/arrowbased/arrowRecordIterator_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/internal/rows/arrowbased/arrowRecordIterator_test.go b/internal/rows/arrowbased/arrowRecordIterator_test.go
index 16a3d8b6..926dcc19 100644
--- a/internal/rows/arrowbased/arrowRecordIterator_test.go
+++ b/internal/rows/arrowbased/arrowRecordIterator_test.go
@@ -193,7 +193,7 @@ func TestArrowRecordIterator(t *testing.T) {
 	})
 }
 
-func TestArrowRecordIteratorSwallowsPageError(t *testing.T) {
+func TestArrowRecordIteratorSurfacesPageErrorOnce(t *testing.T) {
 	logger := dbsqllog.WithContext("connectionId", "correlationId", "")
 
 	executeStatementResp := cli_service.TExecuteStatementResp{}