@@ -29,11 +29,18 @@
 logger = logging.getLogger(__name__)
 
 # Upper bound on how many rows a fetchone() refill pulls and converts at once. The
-# refill uses ``arraysize`` but is capped here so the first fetchone() on a cursor
-# with a large arraysize (default 100000) doesn't materialise a full page just to
-# return one row.
+# refill uses ``arraysize`` but is capped here so a cursor with a large arraysize
+# (default 100000) doesn't materialise a full page just to return one row.
 _FETCHONE_BUFFER_MAX_ROWS = 1000
 
+# fetchone() refills grow geometrically from this size (1 -> 10 -> 100 -> ...) up to
+# min(arraysize, _FETCHONE_BUFFER_MAX_ROWS). The first refill converts a single row,
+# so a lone fetchone() (e.g. SQLAlchemy ``.first()`` / ``.scalar()`` or an EXISTS
+# probe) costs the same as before batching, while iteration loops quickly reach the
+# full batch size.
+_FETCHONE_BUFFER_INITIAL_ROWS = 1
+_FETCHONE_BUFFER_GROWTH_FACTOR = 10
+
 # After a fetchone() batch conversion fails, the batch is retried one row at a time.
 # If this many leading rows all fail too, with the same error type and message as
 # the batch, the failure is treated as systematic: the rest of the batch is not
@@ -124,6 +131,9 @@ def __init__(
         self._row_buffer_pos = 0
         # Set once a fetchone() conversion failure has been logged at WARNING.
         self._fetchone_conversion_warned = False
+        # Size of the next fetchone() refill; grows on each refill (see
+        # ``_FETCHONE_BUFFER_GROWTH_FACTOR``).
+        self._fetchone_refill_rows = _FETCHONE_BUFFER_INITIAL_ROWS
 
     def __iter__(self):
         while True:
@@ -336,8 +346,11 @@ def _fetchone_buffered(
     ) -> Optional[Row]:
         """Serve fetchone() from the row buffer, refilling in small batches.
 
-        Each refill asks ``fetch_raw`` for up to ``arraysize`` rows, capped at
-        ``_FETCHONE_BUFFER_MAX_ROWS``. ``fetch_raw`` may return fewer (Thrift and the
+        Refill sizes start at ``_FETCHONE_BUFFER_INITIAL_ROWS`` and grow by
+        ``_FETCHONE_BUFFER_GROWTH_FACTOR`` on each refill, capped at ``arraysize``
+        and ``_FETCHONE_BUFFER_MAX_ROWS``, so a single fetchone() converts only one
+        row while iteration settles into full batches. ``fetch_raw`` may return
+        fewer (Thrift and the
         kernel backend only serve what is already buffered, refilling at most until
         one row is available, so latency to the first row is unchanged).
 
@@ -346,9 +359,22 @@ def _fetchone_buffered(
         queue is only read again once the buffer is empty, so at that point the
         index equals the queue position again (Thrift relies on that for
         ``expected_row_start_offset``).
+
+        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.
         """
         if self._row_buffer_count() == 0:
-            raw = fetch_raw(max(min(self.arraysize, _FETCHONE_BUFFER_MAX_ROWS), 1))
+            cap = max(min(self.arraysize, _FETCHONE_BUFFER_MAX_ROWS), 1)
+            target = getattr(
+                self, "_fetchone_refill_rows", _FETCHONE_BUFFER_INITIAL_ROWS
+            )
+            size = max(min(target, cap), 1)
+            self._fetchone_refill_rows = min(
+                size * _FETCHONE_BUFFER_GROWTH_FACTOR, _FETCHONE_BUFFER_MAX_ROWS
+            )
+            raw = fetch_raw(size)
             try:
                 rows = convert(raw)
             except Exception as batch_error:
@@ -379,6 +405,9 @@ def _fetchone_buffered(
                 # 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.
+                # The raw chunk is not buffered, so its raw_count rows are lost
+                # (yet counted in rownumber): treat the result set as unusable
+                # after this error; later fetches resume from the next chunk.
                 raise InternalError(
                     "fetchone(): converted {} row(s) from a chunk of {} raw row(s)".format(
                         len(rows), raw_count
 
@@ -370,6 +370,25 @@ def test_mixed_fetches_across_pages_keep_order_and_refill_offsets(self):
         self.assertEqual(rs.rownumber, 10)
         self.assertEqual(offsets, [0, 3, 6, 9])
 
+    def test_close_drops_fetchone_row_buffer(self):
+        # Rows parked in the fetchone() buffer must not survive close(). The warm
+        # refill pulls every row into the buffer, so the queue itself is empty and
+        # any row returned after close() could only come from a stale buffer.
+        from databricks.sql.result_set import _FETCHONE_BUFFER_MAX_ROWS
+
+        for fetch_after_close in (
+            lambda rs: rs.fetchall(),
+            lambda rs: rs.fetchmany_arrow(10).to_pylist(),
+        ):
+            rs = self.make_dummy_result_set_from_initial_results([[1], [2], [3], [4]])
+            rs._fetchone_refill_rows = _FETCHONE_BUFFER_MAX_ROWS
+            self.assertSequenceEqual(rs.fetchone(), [1])
+            self.assertEqual(rs._row_buffer_count(), 3)
+            rs.close()
+            self.assertEqual(rs._row_buffer_count(), 0)
+            self.assertEqual(fetch_after_close(rs), [])
+            self.assertIsNone(rs.fetchone())
+
     def test_fetchone_refill_does_not_fetch_extra_pages(self):
         # A refill must not make extra round-trips to fill ``arraysize`` when
         # server pages are smaller (e.g. byte-limited): it serves what the current
@@ -408,6 +427,10 @@ def test_cloudfetch_fetchone_refill_capped_to_current_chunk(self):
             description=[("col0", "integer", None, None, None, None, None)],
         )
         rs.arraysize = 100
+        # Skip the adaptive refill ramp so the chunk cap, not the ramp, bounds it.
+        from databricks.sql.result_set import _FETCHONE_BUFFER_MAX_ROWS
+
+        rs._fetchone_refill_rows = _FETCHONE_BUFFER_MAX_ROWS
 
         self.assertSequenceEqual(rs.fetchone(), [8])
         queue.next_n_rows.assert_called_once_with(2)
@@ -482,13 +505,21 @@ class ColumnarFetchoneBufferTests(unittest.TestCase):
     rows in batches (not one row at a time) while keeping order and rownumber."""
 
     @staticmethod
-    def make_columnar_result_set(n_rows, arraysize):
+    def make_columnar_result_set(n_rows, arraysize, warm=True):
+        """Build a result set over ``n_rows`` columnar rows.
+
+        With ``warm`` (the default) the adaptive fetchone() refill ramp is skipped,
+        so every refill is ``min(arraysize, _FETCHONE_BUFFER_MAX_ROWS)`` rows and the
+        batch-boundary assertions below stay simple. Pass ``warm=False`` to test
+        the ramp itself.
+        """
+        from databricks.sql.result_set import _FETCHONE_BUFFER_MAX_ROWS
         from databricks.sql.utils import ColumnQueue, ColumnTable
 
         table = ColumnTable([list(range(n_rows))], ["col0"])
         mock_thrift_backend = Mock(spec=ThriftDatabricksClient)
         mock_thrift_backend.fetch_results.return_value = (ColumnQueue(table), False, 0)
-        return ThriftResultSet(
+        rs = ThriftResultSet(
             connection=Mock(),
             execute_response=ExecuteResponse(
                 command_id=None,
@@ -501,6 +532,9 @@ def make_columnar_result_set(n_rows, arraysize):
             thrift_client=mock_thrift_backend,
             arraysize=arraysize,
         )
+        if warm:
+            rs._fetchone_refill_rows = _FETCHONE_BUFFER_MAX_ROWS
+        return rs
 
     @staticmethod
     def _count_conversions():
@@ -543,6 +577,40 @@ def test_fetchone_refill_is_capped_for_large_arraysize(self):
         self.assertEqual([r[0] for r in rs.fetchall()], list(range(1, n_rows)))
         self.assertEqual(rs.rownumber, n_rows)
 
+    def test_first_fetchone_converts_a_single_row(self):
+        # A lone fetchone() (SQLAlchemy .first()/.scalar(), EXISTS probes) must not
+        # pull and convert a whole batch: the first refill is a single row.
+        rs = self.make_columnar_result_set(5000, arraysize=100000, warm=False)
+        with self._count_conversions() as m:
+            self.assertEqual(rs.fetchone()[0], 0)
+        self.assertEqual(m.call_count, 1)
+        self.assertEqual(m.call_args[0][1].num_rows, 1)
+        self.assertEqual(rs._row_buffer_count(), 0)
+        self.assertEqual(rs.rownumber, 1)
+
+    def test_fetchone_refill_size_ramps_up_to_cap(self):
+        from databricks.sql.result_set import _FETCHONE_BUFFER_MAX_ROWS
+
+        n_rows = 1 + 10 + 100 + _FETCHONE_BUFFER_MAX_ROWS * 2
+        rs = self.make_columnar_result_set(n_rows, arraysize=100000, warm=False)
+        with self._count_conversions() as m:
+            rows = [r[0] for r in rs]
+        self.assertEqual(rows, list(range(n_rows)))
+        sizes = [c[0][1].num_rows for c in m.call_args_list]
+        self.assertEqual(
+            sizes[:5],
+            [1, 10, 100, _FETCHONE_BUFFER_MAX_ROWS, _FETCHONE_BUFFER_MAX_ROWS],
+        )
+        self.assertEqual(rs.rownumber, n_rows)
+
+    def test_fetchone_refill_ramp_is_capped_by_arraysize(self):
+        rs = self.make_columnar_result_set(10, arraysize=4, warm=False)
+        with self._count_conversions() as m:
+            rows = [r[0] for r in rs]
+        self.assertEqual(rows, list(range(10)))
+        sizes = [c[0][1].num_rows for c in m.call_args_list]
+        self.assertEqual(sizes[:4], [1, 4, 4, 1])
+
     def test_mixed_fetches_keep_order_and_rownumber(self):
         rs = self.make_columnar_result_set(10, arraysize=4)
         self.assertEqual(rs.fetchone()[0], 0)
 
@@ -308,6 +308,29 @@ def test_close_when_connection_closed(
         assert result_set.has_been_closed_server_side is True
         assert result_set.status == CommandState.CLOSED
 
+    @pytest.mark.parametrize("fetch_after_close", ["fetchall", "fetchmany_arrow"])
+    def test_close_drops_fetchone_row_buffer(
+        self, result_set_with_data, fetch_after_close
+    ):
+        """Rows parked in the fetchone() buffer must not survive close()."""
+        if fetch_after_close == "fetchmany_arrow" and pyarrow is None:
+            pytest.skip("PyArrow is not installed")
+        from databricks.sql.result_set import _FETCHONE_BUFFER_MAX_ROWS
+
+        rs = result_set_with_data
+        # Warm refill pulls all 5 rows into the buffer, emptying the queue, so any
+        # row returned after close() could only come from a stale buffer.
+        rs._fetchone_refill_rows = _FETCHONE_BUFFER_MAX_ROWS
+        assert rs.fetchone()[0] == "value1"
+        assert rs._row_buffer_count() == 4
+        rs.close()
+        assert rs._row_buffer_count() == 0
+        if fetch_after_close == "fetchall":
+            assert rs.fetchall() == []
+        else:
+            assert rs.fetchmany_arrow(10).num_rows == 0
+        assert rs.fetchone() is None
+
     def test_convert_json_types(self, result_set_with_data, sample_data):
         """Test the _convert_json_types method."""
         # Call _convert_json_types
@@ -690,6 +713,10 @@ def test_cloudfetch_fetchone_refill_capped_to_current_chunk(
                 buffer_size_bytes=1000,
                 arraysize=100,
             )
+        # Skip the adaptive refill ramp so the chunk cap, not the ramp, bounds it.
+        from databricks.sql.result_set import _FETCHONE_BUFFER_MAX_ROWS
+
+        rs._fetchone_refill_rows = _FETCHONE_BUFFER_MAX_ROWS
         assert rs.fetchone().col2 == 8
         queue.next_n_rows.assert_called_once_with(2)
         assert rs.fetchone().col2 == 9