stream_to.hxx

Go to the documentation of this file.

/* Definition of the pqxx::stream_to class.
 *
 * pqxx::stream_to enables optimized batch updates to a database table.
 *
 * DO NOT INCLUDE THIS FILE DIRECTLY; include pqxx/stream_to.hxx instead.
 *
 * Copyright (c) 2000-2025, 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_H_STREAM_TO
#define PQXX_H_STREAM_TO
 
#if !defined(PQXX_HEADER_PRE)
#  error "Include libpqxx headers as <pqxx/header>, not <pqxx/header.hxx>."
#endif
 
#include <optional>
#include "pqxx/separated_list.hxx"
#include "pqxx/transaction_base.hxx"
 
 
namespace pqxx
{

class PQXX_LIBEXPORT stream_to : transaction_focus
{
public:

  static stream_to raw_table(
    transaction_base &tx, std::string_view path, std::string_view columns = "")
  {
    return {tx, path, columns};
  }


  static stream_to table(
    transaction_base &tx, table_path path,
    std::initializer_list<std::string_view> columns = {})
  {
    auto const &cx{tx.conn()};
    return raw_table(tx, cx.quote_table(path), cx.quote_columns(columns));
  }
 
#if defined(PQXX_HAVE_CONCEPTS)

  template<PQXX_CHAR_STRINGS_ARG COLUMNS>
  static stream_to
  table(transaction_base &tx, table_path path, COLUMNS const &columns)
  {
    auto const &cx{tx.conn()};
    return stream_to::raw_table(
      tx, cx.quote_table(path), tx.conn().quote_columns(columns));
  }


  template<PQXX_CHAR_STRINGS_ARG COLUMNS>
  static stream_to
  table(transaction_base &tx, std::string_view path, COLUMNS const &columns)
  {
    return stream_to::raw_table(tx, path, tx.conn().quote_columns(columns));
  }
#endif // PQXX_HAVE_CONCEPTS
 
  explicit stream_to(stream_to &&other) :
          // (This first step only moves the transaction_focus base-class
          // object.)
          transaction_focus{std::move(other)},
          m_finished{other.m_finished},
          m_buffer{std::move(other.m_buffer)},
          m_field_buf{std::move(other.m_field_buf)},
          m_finder{other.m_finder}
  {
    other.m_finished = true;
  }
  ~stream_to() noexcept;

  [[nodiscard]] constexpr operator bool() const noexcept
  {
    return not m_finished;
  }
  [[nodiscard]] constexpr bool operator!() const noexcept
  {
    return m_finished;
  }


  void complete();


  template<typename Row> stream_to &operator<<(Row const &row)
  {
    write_row(row);
    return *this;
  }


  stream_to &operator<<(stream_from &);


  template<typename Row> void write_row(Row const &row)
  {
    fill_buffer(row);
    write_buffer();
  }


  template<typename... Ts> void write_values(Ts const &...fields)
  {
    fill_buffer(fields...);
    write_buffer();
  }


  [[deprecated("Use table() or raw_table() factory.")]] stream_to(
    transaction_base &tx, std::string_view table_name) :
          stream_to{tx, table_name, ""sv}
  {}


  template<typename Columns>
  [[deprecated("Use table() or raw_table() factory.")]] stream_to(
    transaction_base &, std::string_view table_name, Columns const &columns);
 
private:
  stream_to(
    transaction_base &tx, std::string_view path, std::string_view columns);
 
  bool m_finished = false;

  std::string m_buffer;

  std::string m_field_buf;

  internal::char_finder_func *m_finder;

  void write_raw_line(std::string_view);


  void write_buffer();

  static constexpr std::string_view null_field{"\\N\t"};

  template<typename T>
  static std::enable_if_t<nullness<T>::always_null, std::size_t>
  estimate_buffer(T const &)
  {
    return std::size(null_field);
  }


  template<typename T>
  static std::enable_if_t<not nullness<T>::always_null, std::size_t>
  estimate_buffer(T const &field)
  {
    return is_null(field) ? std::size(null_field) : size_buffer(field);
  }

  void escape_field_to_buffer(std::string_view data);


  template<typename Field>
  std::enable_if_t<not nullness<Field>::always_null>
  append_to_buffer(Field const &f)
  {
    // We append each field, terminated by a tab.  That will leave us with
    // one tab too many, assuming we write any fields at all; we remove that
    // at the end.
    if (is_null(f))
    {
      // Easy.  Append null and tab in one go.
      m_buffer.append(null_field);
    }
    else
    {
      // Convert f into m_buffer.
 
      using traits = string_traits<Field>;
      auto const budget{estimate_buffer(f)};
      auto const offset{std::size(m_buffer)};
 
      if constexpr (std::is_arithmetic_v<Field>)
      {
        // Specially optimised for "safe" types, which never need any
        // escaping.  Convert straight into m_buffer.
 
        // The budget we get from size_buffer() includes room for the trailing
        // zero, which we must remove.  But we're also inserting tabs between
        // fields, so we re-purpose the extra byte for that.
        auto const total{offset + budget};
        m_buffer.resize(total);
        auto const data{m_buffer.data()};
        char *const end{traits::into_buf(data + offset, data + total, f)};
        *(end - 1) = '\t';
        // Shrink to fit.  Keep the tab though.
        m_buffer.resize(static_cast<std::size_t>(end - data));
      }
      else if constexpr (
        std::is_same_v<Field, std::string> or
        std::is_same_v<Field, std::string_view> or
        std::is_same_v<Field, zview>)
      {
        // This string may need escaping.
        m_field_buf.resize(budget);
        escape_field_to_buffer(f);
      }
      else if constexpr (
        std::is_same_v<Field, std::optional<std::string>> or
        std::is_same_v<Field, std::optional<std::string_view>> or
        std::is_same_v<Field, std::optional<zview>>)
      {
        // Optional string.  It's not null (we checked for that above), so...
        // Treat like a string.
        m_field_buf.resize(budget);
        escape_field_to_buffer(f.value());
      }
      // TODO: Support deleter template argument on unique_ptr.
      else if constexpr (
        std::is_same_v<Field, std::unique_ptr<std::string>> or
        std::is_same_v<Field, std::unique_ptr<std::string_view>> or
        std::is_same_v<Field, std::unique_ptr<zview>> or
        std::is_same_v<Field, std::shared_ptr<std::string>> or
        std::is_same_v<Field, std::shared_ptr<std::string_view>> or
        std::is_same_v<Field, std::shared_ptr<zview>>)
      {
        // TODO: Can we generalise this elegantly without Concepts?
        // Effectively also an optional string.  It's not null (we checked
        // for that above).
        m_field_buf.resize(budget);
        escape_field_to_buffer(*f);
      }
      else
      {
        // This field needs to be converted to a string, and after that,
        // escaped as well.
        m_field_buf.resize(budget);
        auto const data{m_field_buf.data()};
        escape_field_to_buffer(
          traits::to_buf(data, data + std::size(m_field_buf), f));
      }
    }
  }


  template<typename Field>
  std::enable_if_t<nullness<Field>::always_null>
  append_to_buffer(Field const &)
  {
    m_buffer.append(null_field);
  }

  template<typename Container>
  std::enable_if_t<not std::is_same_v<typename Container::value_type, char>>
  fill_buffer(Container const &c)
  {
    // To avoid unnecessary allocations and deallocations, we run through c
    // twice: once to determine how much buffer space we may need, and once to
    // actually write it into the buffer.
    std::size_t budget{0};
    for (auto const &f : c) budget += estimate_buffer(f);
    m_buffer.reserve(budget);
    for (auto const &f : c) append_to_buffer(f);
  }

  template<typename Tuple, std::size_t... indexes>
  static std::size_t
  budget_tuple(Tuple const &t, std::index_sequence<indexes...>)
  {
    return (estimate_buffer(std::get<indexes>(t)) + ...);
  }

  template<typename Tuple, std::size_t... indexes>
  void append_tuple(Tuple const &t, std::index_sequence<indexes...>)
  {
    (append_to_buffer(std::get<indexes>(t)), ...);
  }

  template<typename... Elts> void fill_buffer(std::tuple<Elts...> const &t)
  {
    using indexes = std::make_index_sequence<sizeof...(Elts)>;
 
    m_buffer.reserve(budget_tuple(t, indexes{}));
    append_tuple(t, indexes{});
  }

  template<typename... Ts> void fill_buffer(const Ts &...fields)
  {
    (..., append_to_buffer(fields));
  }
 
  constexpr static std::string_view s_classname{"stream_to"};
};
 
 
template<typename Columns>
inline stream_to::stream_to(
  transaction_base &tx, std::string_view table_name, Columns const &columns) :
        stream_to{tx, table_name, std::begin(columns), std::end(columns)}
{}
} // namespace pqxx
#endif