feat: add server-side cursor for streaming result sets

Pierre-Luc Gagné · claude · Pierre-Luc Gagné · commit 85202ba53885 · 2026-03-31T15:45:14.000-04:00
Add server_cursor&lt;RowType&gt; class that uses MySQL server-side cursors
(CURSOR_TYPE_READ_ONLY) to fetch rows one-at-a-time instead of loading
the entire result set into memory. Created via conn.open_cursor&lt;Row&gt;()
with optional prefetch_rows hint. Includes integration tests.

Co-Authored-By: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -19,6 +19,7 @@ Versioning follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 - `async_query(pool, stmt)` / `async_execute(pool, stmt)` — pool-based async query execution via `std::async`; returns `std::future` holding the result; multiple calls run concurrently on different connections from the pool
 - Composable FK action syntax — two new styles alongside the existing flat types: `fk_attr::on_delete(fk_attr::cascade)` (intermediate) and `fk_attr::on(fk_attr::delete_(fk_attr::cascade))` (fully composed, reads like SQL `ON DELETE CASCADE`); action tags: `cascade`, `restrict_`, `set_null`, `no_action`; all three styles produce identical attribute types
 - `set_case(col, case_when_builder)` — CASE/WHEN expressions in UPDATE SET clauses via `.set_case(col{}, case_when(cond, val).else_(val))` on `update_builder` and `update_set_builder`; chainable with `.set()`, `.where()`, `.order_by()`, `.limit()`
+- `server_cursor<RowType>` — streaming result set via MySQL server-side cursors; created via `conn.open_cursor<RowType>(sql, params...)` or with `prefetch_rows{n}` hint; `.fetch()` returns one row at a time without loading the full result set into memory
 
 ---
 
diff --git a/lib/include/ds_mysql/mysql_connection.hpp b/lib/include/ds_mysql/mysql_connection.hpp
@@ -20,6 +20,7 @@
 #include "ds_mysql/host_name.hpp"
 #include "ds_mysql/port_number.hpp"
 #include "ds_mysql/prepared_statement.hpp"
+#include "ds_mysql/server_cursor.hpp"
 #include "ds_mysql/sql_ddl.hpp"
 #include "ds_mysql/sql_dml.hpp"
 #include "ds_mysql/sql_dql.hpp"
@@ -415,6 +416,48 @@ class mysql_connection {
         return prepare(stmt.build_sql());
     }
 
+    // ---------------------------------------------------------------
+    // Server-side cursors
+    // ---------------------------------------------------------------
+
+    // Open a server-side cursor for streaming large result sets.
+    // Fetches rows one at a time (or in prefetch batches) instead of
+    // loading the entire result set into memory.
+    //
+    //   auto cursor = conn.open_cursor<RowType>("SELECT * FROM t WHERE x = ?", 42u);
+    //   while (auto row = cursor->fetch()) { ... }
+    //
+    // With prefetch hint:
+    //   auto cursor = conn.open_cursor<RowType>(sql, prefetch_rows{100}, 42u);
+    template <typename RowType, typename... Params>
+    [[nodiscard]] std::expected<server_cursor<RowType>, std::string> open_cursor(std::string_view sql,
+                                                                                 Params const&... params) const {
+        return open_cursor_impl<RowType>(sql, prefetch_rows{1}, params...);
+    }
+
+    // Open a server-side cursor with a prefetch_rows hint.
+    template <typename RowType, typename... Params>
+    [[nodiscard]] std::expected<server_cursor<RowType>, std::string> open_cursor(std::string_view sql,
+                                                                                 prefetch_rows hint,
+                                                                                 Params const&... params) const {
+        return open_cursor_impl<RowType>(sql, hint, params...);
+    }
+
+    // Open a server-side cursor from a typed query builder.
+    template <typename RowType, SqlBuilder Stmt, typename... Params>
+    [[nodiscard]] std::expected<server_cursor<RowType>, std::string> open_cursor(Stmt const& stmt,
+                                                                                 Params const&... params) const {
+        return open_cursor_impl<RowType>(stmt.build_sql(), prefetch_rows{1}, params...);
+    }
+
+    // Open a server-side cursor from a typed query builder with prefetch hint.
+    template <typename RowType, SqlBuilder Stmt, typename... Params>
+    [[nodiscard]] std::expected<server_cursor<RowType>, std::string> open_cursor(Stmt const& stmt,
+                                                                                 prefetch_rows hint,
+                                                                                 Params const&... params) const {
+        return open_cursor_impl<RowType>(stmt.build_sql(), hint, params...);
+    }
+
     // Validate that the C++ table struct T matches the live schema in the database.
     //
     // Runs DESCRIBE <table> and checks:
@@ -618,6 +661,60 @@ class mysql_connection {
         return std::string(mysql_error(connection_.get()));
     }
 
+    template <typename RowType, typename... Params>
+    [[nodiscard]] std::expected<server_cursor<RowType>, std::string> open_cursor_impl(std::string_view sql,
+                                                                                       prefetch_rows hint,
+                                                                                       Params const&... params) const {
+        using cursor_t = server_cursor<RowType>;
+        using deleter_t = typename cursor_t::stmt_deleter;
+
+        auto stmt = std::unique_ptr<MYSQL_STMT, deleter_t>(mysql_stmt_init(connection_.get()));
+        if (!stmt) {
+            return std::unexpected(last_error());
+        }
+
+        if (mysql_stmt_prepare(stmt.get(), sql.data(), static_cast<unsigned long>(sql.size())) != 0) {
+            return std::unexpected(std::string(mysql_stmt_error(stmt.get())));
+        }
+
+        // Enable server-side cursor.
+        unsigned long const cursor_type = CURSOR_TYPE_READ_ONLY;
+        if (mysql_stmt_attr_set(stmt.get(), STMT_ATTR_CURSOR_TYPE, &cursor_type)) {
+            return std::unexpected(std::string(mysql_stmt_error(stmt.get())));
+        }
+
+        // Set prefetch rows hint.
+        unsigned long const prefetch = hint.value;
+        if (mysql_stmt_attr_set(stmt.get(), STMT_ATTR_PREFETCH_ROWS, &prefetch)) {
+            return std::unexpected(std::string(mysql_stmt_error(stmt.get())));
+        }
+
+        // Bind input parameters.
+        constexpr auto N = sizeof...(Params);
+        if constexpr (N > 0) {
+            std::array<MYSQL_BIND, N> binds{};
+            std::array<bool, N> null_flags{};
+            auto params_tuple = std::tie(params...);
+            [&]<std::size_t... Is>(std::index_sequence<Is...>) {
+                (stmt_detail::bind_input(binds[Is], std::get<Is>(params_tuple), null_flags[Is]), ...);
+            }(std::make_index_sequence<N>{});
+
+            if (mysql_stmt_bind_param(stmt.get(), binds.data())) {
+                return std::unexpected(std::string(mysql_stmt_error(stmt.get())));
+            }
+        }
+
+        if (mysql_stmt_execute(stmt.get()) != 0) {
+            return std::unexpected(std::string(mysql_stmt_error(stmt.get())));
+        }
+
+        auto cursor = cursor_t{std::move(stmt)};
+        if (auto r = cursor.bind_results(); !r) {
+            return std::unexpected(r.error());
+        }
+        return cursor;
+    }
+
     std::unique_ptr<MYSQL, decltype(&mysql_close)> connection_;
 };
 
diff --git a/lib/include/ds_mysql/server_cursor.hpp b/lib/include/ds_mysql/server_cursor.hpp
@@ -0,0 +1,209 @@
+#pragma once
+
+#include <mysql.h>
+
+#include <cstring>
+#include <expected>
+#include <memory>
+#include <optional>
+#include <string>
+#include <tuple>
+#include <vector>
+
+#include "ds_mysql/prepared_statement.hpp"
+
+namespace ds_mysql {
+
+// ===================================================================
+// server_cursor — streaming result set via MySQL server-side cursors.
+//
+// Unlike prepared_statement::query() which loads all rows into memory,
+// a server_cursor fetches rows one-at-a-time (or in prefetch batches)
+// from the server.  Ideal for large result sets.
+//
+// Created via mysql_connection::open_cursor():
+//   auto cursor = conn.open_cursor<RowType>("SELECT * FROM t WHERE x = ?", 42u);
+//   while (auto row = cursor->fetch()) {
+//       process(*row);
+//   }
+//
+// Prefetch hint (number of rows to buffer at once):
+//   auto cursor = conn.open_cursor<RowType>(sql, prefetch_rows{100}, 42u);
+// ===================================================================
+
+struct prefetch_rows {
+    unsigned long value = 1;
+};
+
+template <typename RowType>
+class server_cursor {
+public:
+    server_cursor(server_cursor const&) = delete;
+    server_cursor& operator=(server_cursor const&) = delete;
+
+    server_cursor(server_cursor&& other) noexcept
+        : stmt_(std::move(other.stmt_)),
+          result_binds_(std::move(other.result_binds_)),
+          lengths_(std::move(other.lengths_)),
+          nulls_(std::move(other.nulls_)),
+          errors_(std::move(other.errors_)),
+          string_bufs_(std::move(other.string_bufs_)),
+          done_(other.done_) {
+        other.done_ = true;
+    }
+
+    server_cursor& operator=(server_cursor&& other) noexcept {
+        if (this != &other) {
+            close();
+            stmt_ = std::move(other.stmt_);
+            result_binds_ = std::move(other.result_binds_);
+            lengths_ = std::move(other.lengths_);
+            nulls_ = std::move(other.nulls_);
+            errors_ = std::move(other.errors_);
+            string_bufs_ = std::move(other.string_bufs_);
+            done_ = other.done_;
+            other.done_ = true;
+        }
+        return *this;
+    }
+
+    ~server_cursor() {
+        close();
+    }
+
+    /// Fetch the next row.  Returns std::nullopt when no more rows.
+    [[nodiscard]] std::expected<std::optional<RowType>, std::string> fetch() {
+        if (done_) {
+            return std::optional<RowType>{std::nullopt};
+        }
+
+        int const status = mysql_stmt_fetch(stmt_.get());
+        if (status == MYSQL_NO_DATA) {
+            done_ = true;
+            return std::optional<RowType>{std::nullopt};
+        }
+        if (status != 0 && status != MYSQL_DATA_TRUNCATED) {
+            return std::unexpected(std::string(mysql_stmt_error(stmt_.get())));
+        }
+
+        return extract_row(std::make_index_sequence<std::tuple_size_v<RowType>>{}, status == MYSQL_DATA_TRUNCATED);
+    }
+
+    /// True when all rows have been fetched.
+    [[nodiscard]] bool done() const noexcept {
+        return done_;
+    }
+
+private:
+    friend class mysql_connection;
+
+    struct stmt_deleter {
+        void operator()(MYSQL_STMT* s) const noexcept {
+            if (s)
+                mysql_stmt_close(s);
+        }
+    };
+
+    static constexpr auto N = std::tuple_size_v<RowType>;
+    static constexpr std::size_t initial_string_buf_size = 256;
+
+    explicit server_cursor(std::unique_ptr<MYSQL_STMT, stmt_deleter> stmt)
+        : stmt_(std::move(stmt)),
+          result_binds_(N),
+          lengths_(N),
+          nulls_(std::make_unique<bool[]>(N)),
+          errors_(std::make_unique<bool[]>(N)),
+          string_bufs_(N) {
+        std::fill_n(nulls_.get(), N, false);
+        std::fill_n(errors_.get(), N, false);
+    }
+
+    void close() {
+        if (stmt_) {
+            mysql_stmt_free_result(stmt_.get());
+        }
+    }
+
+    std::expected<void, std::string> bind_results() {
+        [this]<std::size_t... Is>(std::index_sequence<Is...>) {
+            (setup_bind<std::tuple_element_t<Is, RowType>>(Is), ...);
+        }(std::make_index_sequence<N>{});
+
+        if (mysql_stmt_bind_result(stmt_.get(), result_binds_.data())) {
+            return std::unexpected(std::string(mysql_stmt_error(stmt_.get())));
+        }
+        return {};
+    }
+
+    template <typename T>
+    void setup_bind(std::size_t idx) {
+        auto& bind = result_binds_[idx];
+        std::memset(&bind, 0, sizeof(MYSQL_BIND));
+        bind.length = &lengths_[idx];
+        bind.is_null = &nulls_.get()[idx];
+        bind.error = &errors_.get()[idx];
+
+        using raw = stmt_detail::unwrap_param_type_t<T>;
+        if constexpr (std::same_as<raw, std::string>) {
+            string_bufs_[idx].resize(initial_string_buf_size);
+            bind.buffer_type = MYSQL_TYPE_STRING;
+            bind.buffer = string_bufs_[idx].data();
+            bind.buffer_length = static_cast<unsigned long>(string_bufs_[idx].size());
+        } else {
+            bind.buffer_type = stmt_detail::mysql_type_traits<raw>::field_type;
+            bind.is_unsigned = stmt_detail::mysql_type_traits<raw>::is_unsigned;
+            string_bufs_[idx].resize(sizeof(raw));
+            bind.buffer = string_bufs_[idx].data();
+            bind.buffer_length = sizeof(raw);
+        }
+    }
+
+    template <std::size_t... Is>
+    std::expected<std::optional<RowType>, std::string> extract_row(std::index_sequence<Is...>, bool truncated) {
+        if (truncated) {
+            (refetch_truncated<std::tuple_element_t<Is, RowType>>(Is), ...);
+        }
+        return std::optional<RowType>{
+            RowType{extract_value<std::tuple_element_t<Is, RowType>>(Is)...}};
+    }
+
+    template <typename T>
+    void refetch_truncated(std::size_t idx) {
+        using raw = stmt_detail::unwrap_param_type_t<T>;
+        if constexpr (std::same_as<raw, std::string>) {
+            if (lengths_[idx] > result_binds_[idx].buffer_length) {
+                string_bufs_[idx].resize(lengths_[idx]);
+                result_binds_[idx].buffer = string_bufs_[idx].data();
+                result_binds_[idx].buffer_length = static_cast<unsigned long>(string_bufs_[idx].size());
+                mysql_stmt_fetch_column(stmt_.get(), &result_binds_[idx], static_cast<unsigned int>(idx), 0);
+            }
+        }
+    }
+
+    template <typename T>
+    T extract_value(std::size_t idx) const {
+        if constexpr (is_optional_v<T>) {
+            if (nulls_.get()[idx])
+                return std::nullopt;
+            using inner = unwrap_optional_t<T>;
+            return extract_value<inner>(idx);
+        } else if constexpr (std::same_as<stmt_detail::unwrap_param_type_t<T>, std::string>) {
+            return std::string(string_bufs_[idx].data(), lengths_[idx]);
+        } else {
+            using raw = stmt_detail::unwrap_param_type_t<T>;
+            raw val{};
+            std::memcpy(&val, string_bufs_[idx].data(), sizeof(raw));
+            return static_cast<T>(val);
+        }
+    }
+
+    std::unique_ptr<MYSQL_STMT, stmt_deleter> stmt_;
+    std::vector<MYSQL_BIND> result_binds_;
+    std::vector<unsigned long> lengths_;
+    std::unique_ptr<bool[]> nulls_;
+    std::unique_ptr<bool[]> errors_;
+    std::vector<std::vector<char>> string_bufs_;
+    bool done_ = false;
+};
+
+}  // namespace ds_mysql
diff --git a/tests/integration/mysql/test_query_builder.cpp b/tests/integration/mysql/test_query_builder.cpp
