@@ -34,6 +34,12 @@
 # return one row.
 _FETCHONE_BUFFER_MAX_ROWS = 1000
 
+# After a fetchone() batch conversion fails, the batch is retried one row at a time.
+# If this many leading rows all fail too, the failure is treated as systematic: the
+# rest of the batch is not retried row by row (each would cost a separate
+# conversion) and those rows re-raise the original batch error instead.
+_FETCHONE_FALLBACK_PROBE_ROWS = 5
+
 
 class _FailedRow:
     """Placeholder in the fetchone() row buffer for a row that failed to convert."""
@@ -207,30 +213,44 @@ def _raw_row_count(raw: Any) -> Optional[int]:
 
     @staticmethod
     def _convert_rows_individually(
-        raw: Any, convert: Callable[[Any], List[Row]]
+        raw: Any, convert: Callable[[Any], List[Row]], batch_error: Exception
     ) -> Optional[List[Any]]:
         """Convert ``raw`` row by row after a batch conversion failed.
 
         Rows that fail are returned as ``_FailedRow`` placeholders so the error is
-        raised only when that row is handed out. Returns None if ``raw`` has no rows
-        to split, in which case the caller re-raises the batch error.
+        raised only when that row is handed out. If the first
+        ``_FETCHONE_FALLBACK_PROBE_ROWS`` rows all fail, the failure is treated as
+        systematic and the remaining rows are parked with ``batch_error`` instead of
+        being converted one by one. Returns None if ``raw`` has no rows to split, in
+        which case the caller re-raises the batch error.
         """
         n = ResultSet._raw_row_count(raw)
         if not n:
             return None
         rows: List[Any] = []
+        systematic = False
         for i in range(n):
+            if i == _FETCHONE_FALLBACK_PROBE_ROWS and all(
+                isinstance(r, _FailedRow) for r in rows
+            ):
+                systematic = True
+                rows.extend(_FailedRow(batch_error) for _ in range(n - i))
+                break
             one = raw[i : i + 1] if isinstance(raw, list) else raw.slice(i, 1)
             try:
                 rows.extend(convert(one))
             except Exception as e:
                 rows.append(_FailedRow(e))
         failed = sum(isinstance(r, _FailedRow) for r in rows)
         logger.warning(
-            "fetchone(): failed to convert %d of %d row(s) in a batch; each failing "
+            "fetchone(): failed to convert %d of %d row(s) in a batch%s; each failing "
             "row raises when fetched",
             failed,
             n,
+            " (first %d rows all failed, rest not retried)" % _FETCHONE_FALLBACK_PROBE_ROWS
+            if systematic
+            else "",
+            exc_info=batch_error,
         )
         return rows
 
@@ -270,13 +290,13 @@ def _fetchone_buffered(
             raw = fetch_raw(max(min(self.arraysize, _FETCHONE_BUFFER_MAX_ROWS), 1))
             try:
                 rows = convert(raw)
-            except Exception:
+            except Exception as batch_error:
                 # The queue has already moved past this batch, so it can't be
                 # re-read. Fall back to converting it one row at a time so that,
                 # as before batching, only the row(s) that fail to convert are
                 # lost: each failing row is parked as a _FailedRow and re-raises
                 # its own error when it is reached.
-                fallback = self._convert_rows_individually(raw, convert)
+                fallback = self._convert_rows_individually(raw, convert, batch_error)
                 if fallback is None:
                     raise
                 rows = fallback
 
@@ -600,6 +600,38 @@ def test_fetchone_convert_failure_on_every_row_reraises_per_row(self):
         self.assertEqual(rs._row_buffer_count(), 0)
         self.assertEqual(rs.fetchone()[0], 4)
 
+    def test_fetchone_systematic_convert_failure_stops_per_row_retries(self):
+        from unittest.mock import patch
+        from databricks.sql.result_set import _FETCHONE_FALLBACK_PROBE_ROWS
+
+        batch_error = ValueError("batch failure")
+        calls = []
+
+        def convert(self_, table):
+            calls.append(table.num_rows)
+            if len(calls) == 1:
+                raise batch_error
+            raise ValueError("row failure")
+
+        rs = self.make_columnar_result_set(50, arraysize=20)
+        with patch.object(
+            ThriftResultSet,
+            "_convert_columnar_table",
+            autospec=True,
+            side_effect=convert,
+        ), self.assertLogs("databricks.sql.result_set", level="WARNING") as logs:
+            for i in range(20):
+                with self.assertRaises(ValueError) as ctx:
+                    rs.fetchone()
+                if i >= _FETCHONE_FALLBACK_PROBE_ROWS:
+                    self.assertIs(ctx.exception, batch_error)
+                self.assertEqual(rs.rownumber, i + 1)
+        # One batch attempt plus only the probe rows were converted individually.
+        self.assertEqual(len(calls), 1 + _FETCHONE_FALLBACK_PROBE_ROWS)
+        self.assertTrue(any("20 of 20 row(s)" in m for m in logs.output))
+        self.assertTrue(any("batch failure" in m for m in logs.output))
+        self.assertEqual(rs.fetchone()[0], 20)
+
     def test_fetchone_raw_converted_count_mismatch_fails_loudly(self):
         from unittest.mock import patch
         from databricks.sql.exc import InternalError