@@ -381,10 +381,12 @@ def _fetchone_buffered(
         if self._row_buffer_count() == 0:
             cap = max(min(self.arraysize, _FETCHONE_BUFFER_MAX_ROWS), 1)
             size = max(min(self._fetchone_refill_rows, cap), 1)
+            raw = fetch_raw(size)
+            # Grow the ramp only once the fetch succeeded, so a transient fetch
+            # error (and a retried fetchone()) does not skip a ramp step.
             self._fetchone_refill_rows = min(
                 size * _FETCHONE_BUFFER_GROWTH_FACTOR, _FETCHONE_BUFFER_MAX_ROWS
             )
-            raw = fetch_raw(size)
             try:
                 rows = convert(raw)
             except _NON_DATA_CONVERSION_ERRORS:
@@ -436,7 +438,15 @@ def _fetchone_buffered(
             # that yields [] for a non-empty chunk raises instead of silently
             # ending iteration and losing those rows.
             raw_count = self._raw_row_count(raw)
-            if raw_count is not None and raw_count != len(rows):
+            if raw_count is None:
+                # Every real queue type (Arrow Table, ColumnTable, list) has a row
+                # count, so an unknown chunk type is itself a bug. Fail loudly
+                # rather than skip the drift check and risk rownumber misalignment.
+                raise InternalError(
+                    "fetchone(): cannot determine the row count of a raw chunk of "
+                    "type {}".format(type(raw).__name__)
+                )
+            if raw_count != len(rows):
                 # fetch_raw advanced _next_row_index (and the Thrift fetch offset)
                 # by raw_count; the buffer must line up 1:1 with the raw chunk or
                 # rownumber and fetchmany_arrow slicing would silently drift.
 
@@ -869,6 +869,30 @@ def convert(self_, table):
         self.assertLessEqual(sum(calls), n * (n - 1).bit_length() + n)
         self.assertEqual(rs._row_buffer_count(), n - 1)
 
+    def test_fetchone_failed_fetch_does_not_advance_refill_ramp(self):
+        from unittest.mock import patch
+
+        rs = self.make_columnar_result_set(100, arraysize=100, warm=False)
+        initial = rs._fetchone_refill_rows
+        real_next_n_rows = rs.results.next_n_rows
+        with patch.object(
+            rs.results,
+            "next_n_rows",
+            side_effect=[RuntimeError("transient"), real_next_n_rows],
+        ):
+            with self.assertRaises(RuntimeError):
+                rs.fetchone()
+        # The failed fetch must not grow the ramp, so a retry refills the same size.
+        self.assertEqual(rs._fetchone_refill_rows, initial)
+        sizes = []
+        with patch.object(
+            rs.results,
+            "next_n_rows",
+            side_effect=lambda n: sizes.append(n) or real_next_n_rows(n),
+        ):
+            self.assertEqual(rs.fetchone()[0], 0)
+        self.assertEqual(sizes, [initial])
+
     def test_fetchone_same_message_bad_rows_do_not_drop_later_good_rows(self):
         from unittest.mock import patch
 
@@ -962,6 +986,29 @@ def test_fetchone_empty_conversion_of_nonempty_chunk_fails_loudly(self):
             with self.assertRaises(InternalError):
                 rs.fetchone()
 
+    def test_fetchone_unknown_raw_chunk_type_fails_loudly(self):
+        from unittest.mock import patch
+        from databricks.sql.exc import InternalError
+
+        class _NoRowCount:
+            pass
+
+        rs = self.make_columnar_result_set(10, arraysize=4)
+        with patch.object(
+            ThriftResultSet,
+            "_fetch_available_from_queue",
+            autospec=True,
+            return_value=_NoRowCount(),
+        ), patch.object(
+            ThriftResultSet,
+            "_convert_columnar_table",
+            autospec=True,
+            return_value=[("x",)],
+        ):
+            with self.assertRaises(InternalError) as ctx:
+                rs.fetchone()
+        self.assertIn("_NoRowCount", str(ctx.exception))
+
     @pytest.mark.skipif(pa is None, reason="PyArrow is not installed")
     def test_fetchone_then_fetchall_arrow_keeps_order(self):
         rs = self.make_columnar_result_set(10, arraysize=4)