@@ -41,6 +41,10 @@
 _FETCHONE_BUFFER_INITIAL_ROWS = 1
 _FETCHONE_BUFFER_GROWTH_FACTOR = 10
 
+# Conversion errors that are not caused by row data. fetchone() re-raises these
+# straight away instead of bisecting the batch to find the failing row(s).
+_NON_DATA_CONVERSION_ERRORS = (MemoryError, RecursionError)
+
 
 class _FailedRow:
     """Placeholder in the fetchone() row buffer for a row that failed to convert."""
@@ -379,6 +383,10 @@ def _fetchone_buffered(
             raw = fetch_raw(size)
             try:
                 rows = convert(raw)
+            except _NON_DATA_CONVERSION_ERRORS:
+                # Resource / interpreter failures say nothing about individual
+                # rows; retrying smaller slices would only repeat (or worsen) them.
+                raise
             except Exception as batch_error:
                 # The queue has already moved past this batch, so it can't be
                 # re-read. Fall back to bisecting it down to single rows so that,
@@ -399,6 +407,12 @@ def _fetchone_buffered(
                     raise
                 self._fetchone_conversion_warned = True
                 rows = fallback
+                if all(isinstance(r, _FailedRow) for r in rows):
+                    # Every row failed: likely systemic (a column type or library
+                    # problem that hits each row). Restart the refill ramp so later
+                    # refills convert one row at a time, as before batching, rather
+                    # than paying a full ~2n-conversion bisection on every refill.
+                    self._fetchone_refill_rows = _FETCHONE_BUFFER_INITIAL_ROWS
             # Check for drift before the empty-result early return, so a converter
             # that yields [] for a non-empty chunk raises instead of silently
             # ending iteration and losing those rows.
 
@@ -725,19 +725,66 @@ def test_fetchone_convert_failure_warns_once_per_result_set(self):
             "_convert_columnar_table",
             side_effect=ValueError("bad value"),
         ), self.assertLogs("databricks.sql.result_set", level="DEBUG") as logs:
-            # Three refill batches, every row failing the same way.
+            # One 4-row refill, then (ramp reset after an all-failing batch) eight
+            # 1-row refills, every row failing the same way.
             for _ in range(12):
                 with self.assertRaises(ValueError):
                     rs.fetchone()
         summaries = [
             r for r in logs.records if "failed to convert" in r.getMessage()
         ]
-        self.assertEqual(len(summaries), 3)
+        self.assertEqual(len(summaries), 9)
         self.assertEqual(
             [r.levelno for r in summaries],
-            [logging.WARNING, logging.DEBUG, logging.DEBUG],
+            [logging.WARNING] + [logging.DEBUG] * 8,
         )
 
+    def test_fetchone_all_failing_batch_resets_refill_ramp(self):
+        from unittest.mock import patch
+
+        calls = []
+
+        def convert(self_, table):
+            calls.append(table.num_rows)
+            raise ValueError("systemic failure")
+
+        rs = self.make_columnar_result_set(30, arraysize=10)
+        with patch.object(
+            ThriftResultSet,
+            "_convert_columnar_table",
+            autospec=True,
+            side_effect=convert,
+        ), self.assertLogs("databricks.sql.result_set", level="DEBUG"):
+            for _ in range(15):
+                with self.assertRaises(ValueError):
+                    rs.fetchone()
+        # First full refill is bisected (2 * 10 - 1 conversions); after that every
+        # refill is a single row costing exactly one conversion.
+        self.assertEqual(len(calls), 2 * 10 - 1 + 5)
+        self.assertEqual(calls[-5:], [1] * 5)
+        self.assertEqual(rs.rownumber, 15)
+
+    def test_fetchone_memory_error_is_not_bisected(self):
+        from unittest.mock import patch
+
+        calls = []
+
+        def convert(self_, table):
+            calls.append(table.num_rows)
+            raise MemoryError()
+
+        rs = self.make_columnar_result_set(10, arraysize=4)
+        with patch.object(
+            ThriftResultSet,
+            "_convert_columnar_table",
+            autospec=True,
+            side_effect=convert,
+        ):
+            with self.assertRaises(MemoryError):
+                rs.fetchone()
+        self.assertEqual(calls, [4])
+        self.assertEqual(rs._row_buffer_count(), 0)
+
     def test_fetchone_convert_failure_on_every_row_bisects_to_single_rows(self):
         from unittest.mock import patch