@@ -19,7 +19,7 @@
 from databricks.sql.backend.sea.queue import JsonQueue, SeaResultSetQueueFactory
 from databricks.sql.backend.types import ExecuteResponse
 from databricks.sql.result_set import ResultSet
-from databricks.sql.utils import concat_table_chunks
+from databricks.sql.utils import CloudFetchQueue, concat_table_chunks
 
 logger = logging.getLogger(__name__)
 
@@ -186,6 +186,22 @@ def _fetch_raw_from_queue(self, size: Optional[int]) -> Any:
         self._next_row_index += self._raw_num_rows(results)
         return results
 
+    def _fetch_available_from_queue(self, size: int) -> Any:
+        """Fetch up to ``size`` rows the queue already holds, for Arrow fetchone() refills.
+
+        On CloudFetch, ``next_n_rows`` keeps walking downloaded links until it has
+        ``size`` rows, so a refill that starts near the end of the current chunk
+        would block on the next download before returning a row that is already
+        available. Cap the request at what the current chunk still holds (matching
+        Thrift's ``_fetch_available_from_queue`` and the kernel's ``_fetch_available``).
+        Only called while the fetchone() row buffer is empty.
+        """
+        if isinstance(self.results, CloudFetchQueue) and self.results.table:
+            available = self.results.table.num_rows - self.results.table_row_index
+            if available > 0:
+                size = min(size, available)
+        return self._fetch_raw_from_queue(size)
+
     def _fetch_raw_with_buffer(self, size: Optional[int]) -> Any:
         """Fetch ``size`` (all if None) raw rows, draining the fetchone() buffer first."""
         buffered = self._row_buffer_take_raw(size)
@@ -247,7 +263,9 @@ def fetchone(self) -> Optional[Row]:
 
         if isinstance(self.results, JsonQueue):
             return self._fetchone_buffered(self.fetchmany_json, self._create_json_table)
-        return self._fetchone_buffered(self.fetchmany_arrow, self._convert_arrow_table)
+        return self._fetchone_buffered(
+            self._fetch_available_from_queue, self._convert_arrow_table
+        )
 
     def fetchmany(self, size: int) -> List[Row]:
         """
 
@@ -34,6 +34,15 @@
 _FETCHONE_BUFFER_MAX_ROWS = 1000
 
 
+class _FailedRow:
+    """Placeholder in the fetchone() row buffer for a row that failed to convert."""
+
+    __slots__ = ("error",)
+
+    def __init__(self, error: Exception):
+        self.error = error
+
+
 class ResultSet(ABC):
     """
     Abstract base class for result sets returned by different backend implementations.
@@ -179,7 +188,44 @@ def _row_buffer_take_rows(self, n: Optional[int] = None) -> List[Row]:
         start, length = self._row_buffer_advance(n)
         if self._row_buffer_count() == 0:
             self._row_buffer_clear()
-        return rows[start : start + length]
+        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
+        return taken
+
+    @staticmethod
+    def _convert_rows_individually(
+        raw: Any, convert: Callable[[Any], List[Row]]
+    ) -> 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.
+        """
+        if raw is None:
+            return None
+        n = len(raw) if isinstance(raw, list) else getattr(raw, "num_rows", 0)
+        if not n:
+            return None
+        rows: List[Any] = []
+        for i in range(n):
+            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 "
+            "row raises when fetched",
+            failed,
+            n,
+        )
+        return rows
 
     def _row_buffer_take_raw(self, n: Optional[int] = None) -> Any:
         """Pop up to ``n`` (all if None) un-converted rows from the buffer.
@@ -215,29 +261,18 @@ def _fetchone_buffered(
         """
         if self._row_buffer_count() == 0:
             raw = fetch_raw(max(min(self.arraysize, _FETCHONE_BUFFER_MAX_ROWS), 1))
-            # If convert() raises, this batch is dropped on purpose: the queue has
-            # already moved past it, and _next_row_index (advanced by fetch_raw)
-            # still equals the queue position. Rolling the index back would make it
-            # lag the queue and break Thrift's expected_row_start_offset check on
-            # the next fetch, and retrying convert() on the same data would just
-            # fail again. A warning records how many rows were discarded so a
-            # caller that swallows the error and keeps fetching isn't silently
-            # short of rows.
             try:
                 rows = convert(raw)
             except Exception:
-                if raw is None:
-                    dropped = 0
-                elif isinstance(raw, list):
-                    dropped = len(raw)
-                else:
-                    dropped = getattr(raw, "num_rows", 0)
-                logger.warning(
-                    "fetchone(): failed to convert a batch of %d row(s); these rows "
-                    "are discarded and will not be returned by later fetches",
-                    dropped,
-                )
-                raise
+                # 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)
+                if fallback is None:
+                    raise
+                rows = fallback
             if not rows:
                 return None
             self._row_buffer_raw = raw
 
@@ -534,21 +534,49 @@ def test_mixed_fetches_keep_order_and_rownumber(self):
         self.assertEqual(rs.rownumber, 10)
         self.assertIsNone(rs.fetchone())
 
-    def test_fetchone_convert_failure_logs_dropped_rows_and_reraises(self):
+    def test_fetchone_convert_failure_loses_only_the_bad_row(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
+
         rs = self.make_columnar_result_set(10, arraysize=4)
         with patch.object(
             ThriftResultSet,
             "_convert_columnar_table",
-            side_effect=ValueError("bad value"),
+            autospec=True,
+            side_effect=convert_failing_on_2,
         ), self.assertLogs("databricks.sql.result_set", level="WARNING") as logs:
+            self.assertEqual(rs.fetchone()[0], 0)
+            self.assertEqual(rs.fetchone()[0], 1)
             with self.assertRaises(ValueError):
                 rs.fetchone()
-        self.assertTrue(any("batch of 4 row(s)" in m for m in logs.output))
-        # The failed batch is dropped; the next fetch resumes after it.
+            # Only the failing row is lost; the rest of the batch is still served.
+            self.assertEqual(rs.rownumber, 3)
+            self.assertEqual(rs.fetchone()[0], 3)
+            self.assertEqual([r[0] for r in rs.fetchall()], list(range(4, 10)))
+        self.assertTrue(any("1 of 4 row(s)" in m for m in logs.output))
+        self.assertEqual(rs.rownumber, 10)
+
+    def test_fetchone_convert_failure_on_every_row_reraises_per_row(self):
+        from unittest.mock import patch
+
+        rs = self.make_columnar_result_set(6, arraysize=4)
+        with patch.object(
+            ThriftResultSet,
+            "_convert_columnar_table",
+            side_effect=ValueError("bad value"),
+        ):
+            for expected_rownumber in range(1, 5):
+                with self.assertRaises(ValueError):
+                    rs.fetchone()
+                self.assertEqual(rs.rownumber, expected_rownumber)
         self.assertEqual(rs._row_buffer_count(), 0)
-        self.assertEqual(rs.rownumber, 4)
         self.assertEqual(rs.fetchone()[0], 4)
 
     @pytest.mark.skipif(pa is None, reason="PyArrow is not installed")
 
@@ -658,6 +658,43 @@ def test_arrow_fetchone_then_fetchall_arrow_keeps_order(
         assert table.column(1).to_pylist() == list(range(1, 10))
         assert rs.rownumber == 10
 
+    @pytest.mark.skipif(pyarrow is None, reason="PyArrow is not installed")
+    def test_cloudfetch_fetchone_refill_capped_to_current_chunk(
+        self, mock_connection, mock_sea_client, execute_response
+    ):
+        """fetchone() must not block on the next link while the current chunk has rows."""
+        from databricks.sql.backend.sea.queue import SeaCloudFetchQueue
+
+        chunk = pyarrow.Table.from_pydict(
+            {
+                "col1": ["v{}".format(i) for i in range(10)],
+                "col2": list(range(10)),
+                "col3": [True] * 10,
+            }
+        )
+        queue = Mock(spec=SeaCloudFetchQueue)
+        queue.table = chunk
+        queue.table_row_index = 8
+        queue.next_n_rows.side_effect = lambda n: chunk.slice(8, n)
+        result_data = ResultData(data=None, external_links=[], row_count=10)
+        with patch(
+            "databricks.sql.backend.sea.queue.SeaResultSetQueueFactory.build_queue",
+            return_value=queue,
+        ):
+            rs = SeaResultSet(
+                connection=mock_connection,
+                execute_response=execute_response,
+                sea_client=mock_sea_client,
+                result_data=result_data,
+                manifest=self._create_empty_manifest(ResultFormat.ARROW_STREAM),
+                buffer_size_bytes=1000,
+                arraysize=100,
+            )
+        assert rs.fetchone().col2 == 8
+        queue.next_n_rows.assert_called_once_with(2)
+        assert rs.fetchone().col2 == 9
+        assert rs.rownumber == 2
+
     def test_is_staging_operation(
         self, mock_connection, mock_sea_client, execute_response
     ):