@@ -302,18 +302,21 @@ def convert_range(lo: int, hi: int, error: Optional[Exception]) -> None:
         convert_range(0, n, batch_error)
         failed = sum(isinstance(r, _FailedRow) for r in rows)
         # One summary line, no traceback: the error itself still reaches the caller
-        # when the failing row is fetched. The full traceback is only logged at DEBUG.
+        # when the failing row is fetched. Only the exception type is logged here,
+        # since conversion error messages often quote the offending result value;
+        # the message and full traceback are only logged at DEBUG.
         # ``log_level`` lets the caller log this at WARNING only once per result set.
         logger.log(
             log_level,
             "fetchone(): failed to convert %d of %d row(s) in a batch; each failing "
-            "row raises when fetched (%s: %s)",
+            "row raises when fetched (%s)",
             failed,
             n,
             type(batch_error).__name__,
-            batch_error,
         )
-        logger.debug("fetchone(): batch conversion error", exc_info=batch_error)
+        logger.debug(
+            "fetchone(): batch conversion error: %s", batch_error, exc_info=batch_error
+        )
         return rows
 
     def _cap_to_current_chunk(self, size: int) -> int:
@@ -369,7 +372,11 @@ def _fetchone_buffered(
         If conversion yields a different number of rows than the raw chunk held,
         ``InternalError`` is raised and that whole chunk is dropped (the queue and
         ``rownumber`` have already moved past it). The result set should be
-        considered unusable after that error.
+        considered unusable after that error. The same holds when conversion
+        raises one of ``_NON_DATA_CONVERSION_ERRORS`` (e.g. ``MemoryError``,
+        ``RecursionError``): the error is re-raised and the whole refill chunk
+        (up to ``_FETCHONE_BUFFER_MAX_ROWS`` rows, still counted in ``rownumber``)
+        is lost, so callers should not keep fetching from this result set.
         """
         if self._row_buffer_count() == 0:
             cap = max(min(self.arraysize, _FETCHONE_BUFFER_MAX_ROWS), 1)
@@ -383,6 +390,9 @@ def _fetchone_buffered(
             except _NON_DATA_CONVERSION_ERRORS:
                 # Resource / interpreter failures say nothing about individual
                 # rows; retrying smaller slices would only repeat (or worsen) them.
+                # The queue has already moved past ``raw`` and it is not buffered,
+                # so the whole refill chunk is lost: as with the drift error below,
+                # the result set is unusable after this.
                 raise
             except Exception as batch_error:
                 # The queue has already moved past this batch, so it can't be
 
@@ -738,6 +738,11 @@ def test_fetchone_convert_failure_warns_once_per_result_set(self):
             [r.levelno for r in summaries],
             [logging.WARNING] + [logging.DEBUG] * 8,
         )
+        # The error message may quote result data, so it must stay out of WARNING.
+        for r in logs.records:
+            if r.levelno >= logging.WARNING:
+                self.assertNotIn("bad value", r.getMessage())
+                self.assertIn("ValueError", r.getMessage())
 
     def test_fetchone_all_failing_batch_resets_refill_ramp(self):
         from unittest.mock import patch
@@ -821,7 +826,10 @@ def convert(self_, table):
         self.assertEqual(len(calls), 2 * 20 - 1)
         self.assertEqual(sum(1 for c in calls if c == 1), 20)
         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))
+        # Only the exception type is logged at WARNING; the message (which may quote
+        # result data) is kept out of it.
+        self.assertTrue(any("(ValueError)" in m for m in logs.output))
+        self.assertFalse(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):