@@ -196,20 +196,34 @@ def _row_buffer_advance(self, n: Optional[int]) -> Tuple[int, int]:
         return start, length
 
     def _row_buffer_take_rows(self, n: Optional[int] = None) -> List[Row]:
-        """Pop up to ``n`` (all if None) converted rows from the buffer."""
+        """Pop up to ``n`` (all if None) converted rows from the buffer.
+
+        If the requested range contains a ``_FailedRow``, only the rows up to and
+        including the first failed row are consumed and its error is raised. Rows
+        after it stay in the buffer for the next fetch. The good rows before it in
+        the same call are consumed with the failed fetch (a fetch either returns
+        all of its rows or raises), so ``rownumber`` stays aligned with what is
+        left in the buffer.
+        """
         if self._row_buffer_count() == 0:
             return []
         rows = self._row_buffer_rows
-        start, length = self._row_buffer_advance(n)
+        available = self._row_buffer_count()
+        length = available if n is None else max(0, min(n, available))
+        start = self._row_buffer_pos
+        failed: Optional[_FailedRow] = None
+        for i in range(start, start + length):
+            row = rows[i]
+            if isinstance(row, _FailedRow):
+                failed = row
+                length = i - start + 1
+                break
+        self._row_buffer_advance(length)
         if self._row_buffer_count() == 0:
             self._row_buffer_clear()
-        taken = rows[start : start + length]
-        for row in taken:
-            if isinstance(row, _FailedRow):
-                # The taken rows are consumed either way, matching how a failed
-                # conversion behaved before fetchone() buffering.
-                raise row.error.with_traceback(row.traceback)
-        return taken
+        if failed is not None:
+            raise failed.error.with_traceback(failed.traceback)
+        return rows[start : start + length]
 
     @staticmethod
     def _raw_row_count(raw: Any) -> Optional[int]:
 
@@ -992,10 +992,16 @@ def _assert_rows_and_bound(self, rows, calls, arraysize):
         assert len(rows) == n
         assert [r[0] for r in rows] == list(range(n))
         assert all(r[1] == str(r[0]) for r in rows)
-        # Refills are capped per server batch / CloudFetch chunk, so an exact
-        # ceil(n/arraysize) bound is layout-dependent. Only distinguish "a few"
-        # conversions from "~n" (one per row, issue #981).
-        bound = n // 10
+        # Each refill converts up to min(arraysize, _FETCHONE_BUFFER_MAX_ROWS)
+        # rows, but refills are also capped at a server batch / CloudFetch chunk
+        # boundary, which can add one short refill per boundary. Allow 2x the
+        # ideal refill count plus a small constant for that slack — tight
+        # enough to catch tiny refills, not just one conversion per row (#981).
+        from databricks.sql.result_set import _FETCHONE_BUFFER_MAX_ROWS
+
+        per_refill = max(min(arraysize, _FETCHONE_BUFFER_MAX_ROWS), 1)
+        ideal = -(-n // per_refill)
+        bound = 2 * ideal + 10
         assert calls <= bound, (
             "row conversion called {} times for {} rows (arraysize={}); "
             "expected at most {} (issue #981)".format(calls, n, arraysize, bound)
 
@@ -206,9 +206,15 @@ def _assert_rows_and_bound(rows, calls):
     assert len(rows) == _ROW_BUFFER_N
     assert [r[0] for r in rows] == list(range(_ROW_BUFFER_N))
     assert all(r[1] == str(r[0]) for r in rows)
-    # Server batch boundaries add short refills, so an exact ceil(N/arraysize)
-    # bound is layout-dependent. Only distinguish "a few" from "~N" (issue #981).
-    bound = _ROW_BUFFER_N // 10
+    # Each refill converts up to min(arraysize, _FETCHONE_BUFFER_MAX_ROWS) rows;
+    # server batch boundaries can add one short refill each. Allow 2x the ideal
+    # refill count plus a small constant — tight enough to catch tiny refills,
+    # not just one conversion per row (issue #981).
+    from databricks.sql.result_set import _FETCHONE_BUFFER_MAX_ROWS
+
+    per_refill = max(min(_ROW_BUFFER_ARRAYSIZE, _FETCHONE_BUFFER_MAX_ROWS), 1)
+    ideal = -(-_ROW_BUFFER_N // per_refill)
+    bound = 2 * ideal + 10
     assert calls <= bound, (
         f"_convert_arrow_table called {calls} times for {_ROW_BUFFER_N} rows; "
         f"expected at most {bound} (issue #981)"
 
@@ -585,6 +585,52 @@ def convert_failing_on_2(self_, table):
         self.assertTrue(any("1 of 4 row(s)" in m for m in logs.output))
         self.assertEqual(rs.rownumber, 10)
 
+    def _patch_convert_failing_on_2(self):
+        from unittest.mock import patch
+
+        real_convert = ThriftResultSet._convert_columnar_table
+
+        def convert_failing_on_2(self_, table):
+            rows = real_convert(self_, table)
+            if any(r[0] == 2 for r in rows):
+                raise ValueError("bad value")
+            return rows
+
+        return patch.object(
+            ThriftResultSet,
+            "_convert_columnar_table",
+            autospec=True,
+            side_effect=convert_failing_on_2,
+        )
+
+    def test_fetchmany_over_failed_buffered_row_keeps_rows_after_it(self):
+        rs = self.make_columnar_result_set(10, arraysize=4)
+        with self._patch_convert_failing_on_2(), self.assertLogs(
+            "databricks.sql.result_set", level="WARNING"
+        ):
+            # Buffer now holds rows 1, 2 (failed), 3.
+            self.assertEqual(rs.fetchone()[0], 0)
+            with self.assertRaises(ValueError):
+                rs.fetchmany(3)
+            # Consumed up to and including the failed row; row 3 stays buffered.
+            self.assertEqual(rs.rownumber, 3)
+            self.assertEqual(rs._row_buffer_count(), 1)
+            self.assertEqual([r[0] for r in rs.fetchmany(3)], [3, 4, 5])
+            self.assertEqual([r[0] for r in rs.fetchall()], [6, 7, 8, 9])
+        self.assertEqual(rs.rownumber, 10)
+
+    def test_fetchall_over_failed_buffered_row_keeps_rows_after_it(self):
+        rs = self.make_columnar_result_set(10, arraysize=4)
+        with self._patch_convert_failing_on_2(), self.assertLogs(
+            "databricks.sql.result_set", level="WARNING"
+        ):
+            self.assertEqual(rs.fetchone()[0], 0)
+            with self.assertRaises(ValueError):
+                rs.fetchall()
+            self.assertEqual(rs.rownumber, 3)
+            self.assertEqual([r[0] for r in rs.fetchall()], list(range(3, 10)))
+        self.assertEqual(rs.rownumber, 10)
+
     def test_fetchone_convert_failure_on_every_row_reraises_per_row(self):
         from unittest.mock import patch