@@ -245,17 +245,15 @@ def fetchone(self) -> Optional[Row]:
     def fetchmany(self, size: int) -> List[Row]:
         if size < 0:
             raise ValueError(f"fetchmany size must be >= 0, got {size}")
-        buffered = self._row_buffer_take_rows(size)
-        remaining = size - len(buffered)
-        if remaining == 0:
-            return buffered
-        rest = self._convert_arrow_table(self._fetchmany_arrow_unbuffered(remaining))
-        return buffered + rest if buffered else rest
+        return self._fetch_rows_with_buffer(
+            size,
+            lambda n: self._convert_arrow_table(self._fetchmany_arrow_unbuffered(n)),
+        )
 
     def fetchall(self) -> List[Row]:
-        buffered = self._row_buffer_take_rows()
-        rest = self._convert_arrow_table(self._drain())
-        return buffered + rest if buffered else rest
+        return self._fetch_rows_with_buffer(
+            None, lambda _: self._convert_arrow_table(self._drain())
+        )
 
     def close(self) -> None:
         """Close the underlying kernel handle and notify the backend.
 
@@ -279,15 +279,13 @@ def fetchmany(self, size: int) -> List[Row]:
 
         if size < 0:
             raise ValueError(f"size argument for fetchmany is {size} but must be >= 0")
-        buffered = self._row_buffer_take_rows(size)
-        remaining = size - len(buffered)
-        if buffered and remaining == 0:
-            return buffered
         if isinstance(self.results, JsonQueue):
-            rest = self._create_json_table(self.fetchmany_json(remaining))
-        else:
-            rest = self._convert_arrow_table(self.fetchmany_arrow(remaining))
-        return buffered + rest if buffered else rest
+            return self._fetch_rows_with_buffer(
+                size, lambda n: self._create_json_table(self.fetchmany_json(n))
+            )
+        return self._fetch_rows_with_buffer(
+            size, lambda n: self._convert_arrow_table(self.fetchmany_arrow(n))
+        )
 
     def fetchall(self) -> List[Row]:
         """
@@ -297,9 +295,10 @@ def fetchall(self) -> List[Row]:
             List of Row objects containing all remaining rows
         """
 
-        buffered = self._row_buffer_take_rows()
         if isinstance(self.results, JsonQueue):
-            rest = self._create_json_table(self.fetchall_json())
-        else:
-            rest = self._convert_arrow_table(self.fetchall_arrow())
-        return buffered + rest if buffered else rest
+            return self._fetch_rows_with_buffer(
+                None, lambda _: self._create_json_table(self.fetchall_json())
+            )
+        return self._fetch_rows_with_buffer(
+            None, lambda _: self._convert_arrow_table(self.fetchall_arrow())
+        )
@@ -41,6 +41,7 @@
 _FETCHONE_BUFFER_INITIAL_ROWS = 1
 _FETCHONE_BUFFER_GROWTH_FACTOR = 10
 
+
 class _FailedRow:
     """Placeholder in the fetchone() row buffer for a row that failed to convert."""
 
@@ -226,6 +227,26 @@ def _row_buffer_take_rows(self, n: Optional[int] = None) -> List[Row]:
             raise failed.error.with_traceback(failed.traceback)
         return rows[start : start + length]
 
+    def _fetch_rows_with_buffer(
+        self,
+        size: Optional[int],
+        fetch_rest: Callable[[Optional[int]], List[Row]],
+    ) -> List[Row]:
+        """Return ``size`` rows (all if None), draining the fetchone() buffer first.
+
+        Every row-level fetchmany()/fetchall() goes through here, so rows the
+        fetchone() buffer already holds are always returned before newly fetched
+        ones. ``fetch_rest(n)`` must fetch up to ``n`` (all if None) rows
+        while bypassing the buffer. It is not called when the buffer already
+        covers ``size``.
+        """
+        buffered = self._row_buffer_take_rows(size)
+        remaining = None if size is None else size - len(buffered)
+        if buffered and remaining == 0:
+            return buffered
+        rest = fetch_rest(remaining)
+        return buffered + rest if buffered else rest
+
     @staticmethod
     def _raw_row_count(raw: Any) -> Optional[int]:
         """Row count of a raw chunk (Arrow table, ColumnTable or list); None if unknown."""
@@ -774,12 +795,13 @@ def fetchall(self) -> List[Row]:
         """
         Fetch all (remaining) rows of a query result, returning them as a list of rows.
         """
-        buffered = self._row_buffer_take_rows()
         if isinstance(self.results, ColumnQueue):
-            rest = self._convert_columnar_table(self.fetchall_columnar())
-        else:
-            rest = self._convert_arrow_table(self.fetchall_arrow())
-        return buffered + rest if buffered else rest
+            return self._fetch_rows_with_buffer(
+                None, lambda _: self._convert_columnar_table(self.fetchall_columnar())
+            )
+        return self._fetch_rows_with_buffer(
+            None, lambda _: self._convert_arrow_table(self.fetchall_arrow())
+        )
 
     def fetchmany(self, size: int) -> List[Row]:
         """
@@ -789,15 +811,13 @@ def fetchmany(self, size: int) -> List[Row]:
         """
         if size < 0:
             raise ValueError(f"size argument for fetchmany is {size} but must be >= 0")
-        buffered = self._row_buffer_take_rows(size)
-        remaining = size - len(buffered)
-        if buffered and remaining == 0:
-            return buffered
         if isinstance(self.results, ColumnQueue):
-            rest = self._convert_columnar_table(self.fetchmany_columnar(remaining))
-        else:
-            rest = self._convert_arrow_table(self.fetchmany_arrow(remaining))
-        return buffered + rest if buffered else rest
+            return self._fetch_rows_with_buffer(
+                size, lambda n: self._convert_columnar_table(self.fetchmany_columnar(n))
+            )
+        return self._fetch_rows_with_buffer(
+            size, lambda n: self._convert_arrow_table(self.fetchmany_arrow(n))
+        )
 
     @staticmethod
     def _get_schema_description(table_schema_message):
 
@@ -777,6 +777,43 @@ def convert(self_, table):
         self.assertTrue(any("batch failure" in m for m in logs.output))
         self.assertEqual(rs.fetchone()[0], 20)
 
+    def test_fetchone_all_failing_max_batch_conversion_count_is_bounded(self):
+        """Document the worst-case cost of bisecting a full all-failing refill.
+
+        When every row fails (a systemic converter / type problem), bisection tries
+        every node of the tree: exactly ``2n - 1`` conversions for an ``n``-row
+        refill, i.e. about 2x the one-conversion-per-row cost of fetchone() before
+        batching. Short-circuiting on "same exception type in both halves" is
+        deliberately not done, since it would drop good rows that share the error
+        type (see test_fetchone_same_message_bad_rows_do_not_drop_later_good_rows).
+        """
+        from unittest.mock import patch
+        from databricks.sql.result_set import _FETCHONE_BUFFER_MAX_ROWS
+
+        n = _FETCHONE_BUFFER_MAX_ROWS
+        calls = []
+
+        def convert(self_, table):
+            calls.append(table.num_rows)
+            raise ValueError("systemic failure")
+
+        rs = self.make_columnar_result_set(n, arraysize=n)
+        # Skip the adaptive refill ramp so the first refill is a full batch.
+        rs._fetchone_refill_rows = n
+        with patch.object(
+            ThriftResultSet,
+            "_convert_columnar_table",
+            autospec=True,
+            side_effect=convert,
+        ), self.assertLogs("databricks.sql.result_set", level="WARNING"):
+            with self.assertRaises(ValueError):
+                rs.fetchone()
+        self.assertEqual(calls[0], n)
+        self.assertEqual(len(calls), 2 * n - 1)
+        # Total rows converted across all attempts is O(n log n), not O(n^2).
+        self.assertLessEqual(sum(calls), n * (n - 1).bit_length() + n)
+        self.assertEqual(rs._row_buffer_count(), n - 1)
+
     def test_fetchone_same_message_bad_rows_do_not_drop_later_good_rows(self):
         from unittest.mock import patch