stream_from.hxx

Go to the documentation of this file.

/* Definition of the pqxx::stream_from class.
 *
 * pqxx::stream_from enables optimized batch reads from a database table.
 *
 * DO NOT INCLUDE THIS FILE DIRECTLY; include pqxx/stream_from instead.
 *
 * Copyright (c) 2000-2026, Jeroen T. Vermeulen.
 *
 * See COPYING for copyright license.  If you did not receive a file called
 * COPYING with this source code, please notify the distributor of this
 * mistake, or contact the author.
 */
#ifndef PQXX_STREAM_FROM_HXX
#define PQXX_STREAM_FROM_HXX
 
#if !defined(PQXX_HEADER_PRE)
#  error "Include libpqxx headers as <pqxx/header>, not <pqxx/header.hxx>."
#endif
 
#include <cassert>
#include <format>
#include <variant>
 
#include "pqxx/connection.hxx"
#include "pqxx/encoding_group.hxx"
#include "pqxx/except.hxx"
#include "pqxx/internal/stream_iterator.hxx"
#include "pqxx/separated_list.hxx"
#include "pqxx/transaction_focus.hxx"
 
 
namespace pqxx
{
class transaction_base;
 
 
 
constexpr from_table_t from_table;
 
constexpr from_query_t from_query;
 
 
 
class PQXX_LIBEXPORT stream_from final : transaction_focus
{
public:
  using raw_line =
    std::pair<std::unique_ptr<char[], void (*)(void const *)>, std::size_t>;
 
 
  [[deprecated("Use transaction_base::stream instead.")]] static stream_from
  query(transaction_base &tx, std::string_view q)
  {
#include "pqxx/internal/ignore-deprecated-pre.hxx"
    return {tx, from_query, q};
#include "pqxx/internal/ignore-deprecated-post.hxx"
  }
 
 
 
  [[deprecated("Use transaction_base::stream instead.")]] static stream_from
  raw_table(
    transaction_base &tx, std::string_view path,
    std::string_view columns = ""sv);
 
 
  [[deprecated("Use transaction_base::stream instead.")]] static stream_from
  table(
    transaction_base &tx, table_path path,
    std::initializer_list<std::string_view> columns = {});
 
 
  [[deprecated("Use transaction_base::stream instead.")]] stream_from(
    transaction_base &, from_query_t, std::string_view query);
 
 
  [[deprecated("Use transaction_base::stream instead.")]] stream_from(
    transaction_base &, from_table_t, std::string_view table);
 
 
  template<typename Iter>
  [[deprecated("Use transaction_base::stream instead.")]] stream_from(
    transaction_base &, from_table_t, std::string_view table,
    Iter columns_begin, Iter columns_end);
 
 
  template<typename Columns>
  [[deprecated("Use transaction_base::stream() instead.")]] stream_from(
    transaction_base &tx, from_table_t, std::string_view table,
    Columns const &columns);
 
#include "pqxx/internal/ignore-deprecated-pre.hxx"
  [[deprecated("Use transaction_base::stream instead.")]] stream_from(
    transaction_base &tx, std::string_view table) :
          stream_from{tx, from_table, table}
  {}
#include "pqxx/internal/ignore-deprecated-post.hxx"
 
  template<typename Columns>
  [[deprecated("Use transaction_base::stream instead.")]] stream_from(
    transaction_base &tx, std::string_view table, Columns const &columns) :
          stream_from{tx, from_table, table, columns}
  {}
 
  template<typename Iter>
  [[deprecated("Use transaction_base::stream instead.")]] stream_from(
    transaction_base &, std::string_view table, Iter columns_begin,
    Iter columns_end);
 
  stream_from(stream_from const &) = delete;
  stream_from(stream_from &&) = delete;
  ~stream_from() noexcept;
 
  stream_from &operator=(stream_from const &) = delete;
  stream_from &operator=(stream_from &&) = delete;
 
  // NOLINTBEGIN(google-explicit-constructor,hicpp-explicit-conversions)
  // LCOV_EXCL_START
  [[nodiscard]] constexpr operator bool() const noexcept
  {
    return not m_finished;
  }
  // LCOV_EXCL_STOP
  // NOLINTEND(google-explicit-constructor,hicpp-explicit-conversions)
 
  [[nodiscard]] constexpr bool operator!() const noexcept
  {
    return m_finished;
  }
 
 
  void complete(sl = sl::current());
 
 
  template<typename Tuple> stream_from &operator>>(Tuple &);
 
  template<typename... Vs>
  stream_from &operator>>(std::variant<Vs...> &) = delete;
 
 
  template<typename... TYPE> [[nodiscard]] auto iter() &
  {
    return pqxx::internal::stream_input_iteration<TYPE...>{*this};
  }
 
 
  std::vector<std::string_view> const *read_row(sl loc = sl::current()) &;
 
 
  raw_line get_raw_line(sl);
 
private:
  // TODO: Clean up this signature once we cull the deprecated constructors.
  stream_from(
    transaction_base &tx, std::string_view table, std::string_view columns,
    from_table_t);
 
  // TODO: Clean up this signature once we cull the deprecated constructors.
  stream_from(
    transaction_base &, std::string_view unquoted_table,
    std::string_view columns, from_table_t, int);
 
  template<typename Tuple, std::size_t... indexes>
  void extract_fields(Tuple &t, std::index_sequence<indexes...>, sl loc) const
  {
    (extract_value<Tuple, indexes>(t, loc), ...);
  }
 
  pqxx::internal::char_finder_func *m_char_finder;
 
  std::string m_row;
 
  std::vector<std::string_view> m_fields;
 
  bool m_finished = false;
 
  void close();
 
  template<typename Tuple, std::size_t index>
  void extract_value(Tuple &, sl loc) const;
 
  void parse_line(sl);
};
 
 
template<typename Columns>
inline stream_from::stream_from(
  transaction_base &tx, from_table_t, std::string_view table_name,
  Columns const &columns) :
        stream_from{
          tx, from_table, table_name, std::begin(columns), std::end(columns)}
{}
 
 
template<typename Iter>
inline stream_from::stream_from(
  transaction_base &tx, from_table_t, std::string_view table,
  Iter columns_begin, Iter columns_end) :
        stream_from{
          tx, table, separated_list(",", columns_begin, columns_end),
          from_table, 1}
{}
 
 
template<typename Tuple> inline stream_from &stream_from::operator>>(Tuple &t)
{
  sl const loc{sl::current()};
  if (m_finished) [[unlikely]]
    return *this;
  static constexpr auto tup_size{std::tuple_size_v<Tuple>};
  m_fields.reserve(tup_size);
  parse_line(loc);
  if (m_finished) [[unlikely]]
    return *this;
 
  if (std::size(m_fields) != tup_size)
    throw usage_error{std::format(
      "Tried to extract {} field(s) from a stream of {}.", tup_size,
      std::size(m_fields))};
 
  extract_fields(t, std::make_index_sequence<tup_size>{}, loc);
  return *this;
}
 
 
template<typename Tuple, std::size_t index>
inline void stream_from::extract_value(Tuple &t, sl loc) const
{
  using field_type = std::remove_cvref_t<decltype(std::get<index>(t))>;
  assert(index < std::size(m_fields));
  if constexpr (always_null<field_type>())
  {
    if (std::data(m_fields[index]) != nullptr)
      throw conversion_error{"Streaming non-null value into null field.", loc};
  }
  else if (std::data(m_fields[index]) == nullptr)
  {
    if constexpr (has_null<field_type>())
      std::get<index>(t) = make_null<field_type>();
    else
      internal::throw_null_conversion(name_type<field_type>(), loc);
  }
  else
  {
    // Don't ever try to convert a non-null value to nullptr_t!
    std::get<index>(t) = from_string<field_type>(m_fields[index]);
  }
}
} // namespace pqxx
#endif