def dataset_select_paginated(
        self,
        query,
        page_size: int = SELECT_BATCH_SIZE,
    ) -> Generator[Sequence, None, None]:
        """
        This is equivalent to `db.execute`, but for selecting rows in batches
        """
        limit = query._limit
        paginated_query = query.limit(page_size)

        offset = 0
        num_yielded = 0

        # Ensure we're using a thread-local connection
        with self.clone() as wh:
            while True:
                if limit is not None:
                    limit -= num_yielded
                    if limit == 0:
                        break
                    if limit < page_size:
                        paginated_query = paginated_query.limit(None).limit(limit)

                # Cursor results are not thread-safe, so we convert them to a list
                results = list(wh.dataset_rows_select(paginated_query.offset(offset)))

                processed = False
                for row in results:
                    processed = True
                    yield row
                    num_yielded += 1

                if not processed:
                    break  # no more results
                offset += page_size

        This is equivalent to `db.execute`, but for selecting rows in batches
        
src/datachain/data_storage/warehouse.py