postgresql_streaming_source.hpp

Go to the documentation of this file.

#pragma once
 
#include "../results/streaming_result.hpp"
#include "postgresql_connection.hpp"
 
#include <memory>
#include <optional>
#include <string>
#include <vector>
 
// Forward declarations
struct pg_conn;
using PGconn = pg_conn;
 
namespace relx::connection {
 
class PostgreSQLStreamingSource {
public:
  PostgreSQLStreamingSource(PostgreSQLConnection& connection, std::string sql,
                            std::vector<std::string> params = {});
 
  PostgreSQLStreamingSource(PostgreSQLConnection& connection, std::string sql,
                            std::vector<std::string> params, std::vector<bool> is_binary);
 
  ~PostgreSQLStreamingSource();
 
  // Disable copy operations
  PostgreSQLStreamingSource(const PostgreSQLStreamingSource&) = delete;
  PostgreSQLStreamingSource& operator=(const PostgreSQLStreamingSource&) = delete;
 
  // Allow move operations
  PostgreSQLStreamingSource(PostgreSQLStreamingSource&& other) noexcept;
  PostgreSQLStreamingSource& operator=(PostgreSQLStreamingSource&& other) noexcept;
 
  ConnectionResult<void> initialize();
 
  std::optional<std::string> get_next_row();
 
  const std::vector<std::string>& get_column_names() const;
 
  bool is_initialized() const { return initialized_; }
 
  bool has_more_rows() const { return !finished_; }
 
private:
  PostgreSQLConnection& connection_;
  std::string sql_;
  std::vector<std::string> params_;
  std::vector<bool> is_binary_;
  bool use_binary_;
 
  std::vector<std::string> column_names_;
  std::vector<bool> is_bytea_column_;
  bool initialized_;
  bool finished_;
  bool convert_bytea_;
 
  // We need to track the active query state
  bool query_active_;
 
  // Cache for the first row (since we consume it during metadata processing)
  std::optional<std::string> first_row_cached_;
 
  ConnectionResult<void> start_query();
 
  void process_column_metadata(struct pg_result* pg_result);
 
  std::optional<std::string> format_row(struct pg_result* pg_result);
 
  std::string convert_pg_bytea_to_binary(const std::string& hex_value) const;
 
  void cleanup();
};
 
template <typename... Args>
result::StreamingResultSet<PostgreSQLStreamingSource> create_streaming_result(
    PostgreSQLConnection& connection, const std::string& sql, Args&&... args) {
  // Convert parameters to strings (similar to execute_typed)
  std::vector<std::string> param_strings;
  if constexpr (sizeof...(Args) > 0) {
    param_strings.reserve(sizeof...(Args));
 
    auto add_param = [&param_strings](auto&& param) {
      using ParamType = std::remove_cvref_t<decltype(param)>;
 
      if constexpr (std::is_same_v<ParamType, std::nullptr_t>) {
        param_strings.push_back("NULL");
      } else if constexpr (std::is_same_v<ParamType, std::string> ||
                           std::is_same_v<ParamType, const char*> ||
                           std::is_same_v<ParamType, std::string_view>) {
        param_strings.push_back(std::string(param));
      } else if constexpr (std::is_arithmetic_v<ParamType>) {
        param_strings.push_back(std::to_string(param));
      } else if constexpr (std::is_same_v<ParamType, bool>) {
        param_strings.push_back(param ? "t" : "f");
      } else {
        std::ostringstream ss;
        ss << param;
        param_strings.push_back(ss.str());
      }
    };
 
    (add_param(std::forward<Args>(args)), ...);
  }
 
  return result::StreamingResultSet<PostgreSQLStreamingSource>(
      PostgreSQLStreamingSource(connection, sql, std::move(param_strings)));
}
 
}  // namespace relx::connection