From 0ff63871e59384bedb8b20ae3306ddd350390091 Mon Sep 17 00:00:00 2001
From: Leijurv <leijurv@gmail.com>
Date: Sun, 25 Oct 2026 00:11:00 -0700
Subject: [PATCH] Use binary COPY for flex and middle tables

Switch the COPY for the middle tables and most flex tables from the text
format to the binary format, saving the CPU time otherwise spent on
int-to-text-to-int, float-to-text-to-float, WKB hex, etc.

Each db_target_descr_t knows the binary versions of its column types; if
present, db_copy_mgr_t switches to binary format for the same calls
(add_column, new_array, ...) and checks each field's type against the
list. Middle tables always use binary mode (including int8[]). Flex
tables use binary mode if all their columns have supported types and
fall back to text mode otherwise (user-defined SQL types, or
timestamp/timestamptz given from Lua as arbitrary strings). The pgsql
output and gen are unchanged.

Fixes #2286
---
 src/db-copy-mgr.hpp        | 379 ++++++++++++++++++++++++++++++++++++-
 src/db-copy.cpp            |  16 ++
 src/db-copy.hpp            |  50 ++++-
 src/flex-table-column.cpp  |  49 +++++
 src/flex-table-column.hpp  |  11 ++
 src/flex-table.cpp         |  29 +++
 src/flex-table.hpp         |   7 +
 src/middle-pgsql.cpp       |  26 ++-
 tests/test-db-copy-mgr.cpp | 255 +++++++++++++++++++++++++
 9 files changed, 811 insertions(+), 11 deletions(-)

diff --git a/src/db-copy-mgr.hpp b/src/db-copy-mgr.hpp
index a92af7388..7209ba5ce 100644
--- a/src/db-copy-mgr.hpp
+++ b/src/db-copy-mgr.hpp
@@ -10,15 +10,53 @@
  * For a full list of authors see the git log.
  */
 
+#include <array>
 #include <cassert>
+#include <cmath>
+#include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <limits>
 #include <memory>
+#include <stdexcept>
 #include <string>
+#include <string_view>
+#include <type_traits>
+#include <utility>
+
+#include <osmium/osm/timestamp.hpp>
 
 #include "db-copy.hpp"
+#include "format.hpp"
 #include "hex.hpp"
 
+/**
+ * Convert a double to a float for a real column. Values that PostgreSQL
+ * would reject as out of range for type real in the text format (overflow
+ * to infinity or underflow to zero) throw an exception.
+ */
+inline float copy_to_float4(double value)
+{
+    if (std::isnan(value)) { // the text format sends every NaN as "nan"
+        return std::numeric_limits<float>::quiet_NaN();
+    }
+
+    auto const result = static_cast<float>(value);
+    if (std::isfinite(value) &&
+        (std::isinf(result) || (result == 0.0F && value != 0.0))) {
+        throw fmt_error("Value {} is out of range for type real.", value);
+    }
+
+    return result;
+}
+
 /**
  * Management class that fills and manages copy buffers.
+ *
+ * Rows are written in PostgreSQL's text COPY format, or in its binary COPY
+ * format if the target has binary_types() set. The same calls are used for
+ * both, in the binary format the type of each field is taken from the
+ * target's list of column types.
  */
 template <typename DELETER>
 class db_copy_mgr_t
@@ -43,12 +81,15 @@ class db_copy_mgr_t
             m_current = db_cmd_copy_delete_t<DELETER>(table);
         }
         m_committed = m_current.buffer.size();
+        m_binary_types = table->binary() ? &table->binary_types() : nullptr;
+        m_field = 0;
     }
 
     void rollback_line()
     {
         assert(m_current);
         m_current.buffer.resize(m_committed);
+        m_field = 0;
     }
 
     /**
@@ -64,10 +105,15 @@ class db_copy_mgr_t
         auto &buf = m_current.buffer;
         assert(!buf.empty());
 
-        // Expect that a column has been written last which ended in a '\t'.
-        // Replace it with the row delimiter '\n'.
-        assert(buf.back() == '\t');
-        buf.back() = '\n';
+        if (m_binary_types) {
+            // Binary rows have no delimiter, but all fields must be there.
+            assert(m_field == m_binary_types->size());
+        } else {
+            // Expect that a column has been written last which ended in a
+            // '\t'. Replace it with the row delimiter '\n'.
+            assert(buf.back() == '\t');
+            buf.back() = '\n';
+        }
 
         if (m_current.is_full()) {
             m_processor->send_command(std::move(m_current));
@@ -95,6 +141,10 @@ class db_copy_mgr_t
     template <typename T>
     void add_column(T &&value)
     {
+        if (m_binary_types) {
+            add_binary(next_field(), std::forward<T>(value));
+            return;
+        }
         add_value(std::forward<T>(value));
         m_current.buffer += '\t';
     }
@@ -104,7 +154,15 @@ class db_copy_mgr_t
      *
      * Adds a NULL value for the column.
      */
-    void add_null_column() { m_current.buffer += "\\N\t"; }
+    void add_null_column()
+    {
+        if (m_binary_types) {
+            next_field();
+            put_int32(-1); // NULL
+            return;
+        }
+        m_current.buffer += "\\N\t";
+    }
 
     /**
      * Start an array column.
@@ -113,7 +171,20 @@ class db_copy_mgr_t
      *
      * Must be finished with a call to finish_array().
      */
-    void new_array() { m_current.buffer += "{"; }
+    void new_array()
+    {
+        if (m_binary_types) {
+            check_field_type(next_field(), copy_field_type::int8_array);
+            m_field_start = start_field_length();
+            put_int32(1);       // number of dimensions
+            put_int32(0);       // no NULL elements
+            put_int32(INT8OID); // element type
+            put_int32(0);       // length, set in finish_array()
+            put_int32(1);       // lower bound
+            return;
+        }
+        m_current.buffer += "{";
+    }
 
     /**
      * Add a single value to an array column.
@@ -123,6 +194,10 @@ class db_copy_mgr_t
      */
     void add_array_elem(osmid_t value)
     {
+        if (m_binary_types) {
+            put_value(value);
+            return;
+        }
         add_value(value);
         m_current.buffer += ',';
     }
@@ -135,6 +210,21 @@ class db_copy_mgr_t
      */
     void finish_array()
     {
+        if (m_binary_types) {
+            auto const header = m_field_start + 4;
+            auto const elements =
+                (m_current.buffer.size() - header - ARRAY_HEADER_SIZE) /
+                ARRAY_ELEMENT_SIZE;
+            if (elements == 0) {
+                // An empty array has zero dimensions and no dimension info.
+                m_current.buffer.resize(header + 12);
+                put_length_at(header, 0); // number of dimensions
+            } else {
+                put_length_at(header + 12, elements);
+            }
+            finish_field_length(m_field_start);
+            return;
+        }
         assert(!m_current.buffer.empty());
         if (m_current.buffer.back() == '{') {
             m_current.buffer += '}';
@@ -156,7 +246,13 @@ class db_copy_mgr_t
      * Must be closed with a finish_hash() call.
      */
     void new_hash()
-    { /* nothing */
+    {
+        if (m_binary_types) {
+            check_field_type(next_field(), copy_field_type::hstore);
+            m_field_start = start_field_length();
+            put_int32(0); // pairs, set in finish_hash()
+            m_hash_pairs = 0;
+        }
     }
 
     void add_hash_elem(std::string const &k, std::string const &v)
@@ -172,6 +268,10 @@ class db_copy_mgr_t
      */
     void add_hash_elem(char const *k, char const *v)
     {
+        if (m_binary_types) {
+            add_binary_hash_elem(k, v);
+            return;
+        }
         m_current.buffer += '"';
         add_escaped_string(k);
         m_current.buffer += "\"=>\"";
@@ -187,6 +287,10 @@ class db_copy_mgr_t
      */
     void add_hash_elem_noescape(char const *k, char const *v)
     {
+        if (m_binary_types) {
+            add_binary_hash_elem(k, v);
+            return;
+        }
         m_current.buffer += '"';
         m_current.buffer += k;
         m_current.buffer += "\"=>\"";
@@ -207,6 +311,10 @@ class db_copy_mgr_t
     template <typename T>
     void add_hstore_num_noescape(char const *k, T const value)
     {
+        if (m_binary_types) {
+            add_binary_hash_elem(k, std::to_string(value).c_str());
+            return;
+        }
         m_current.buffer += '"';
         m_current.buffer += k;
         m_current.buffer += "\"=>\"";
@@ -222,6 +330,11 @@ class db_copy_mgr_t
      */
     void finish_hash()
     {
+        if (m_binary_types) {
+            put_length_at(m_field_start + 4, m_hash_pairs);
+            finish_field_length(m_field_start);
+            return;
+        }
         auto const idx = m_current.buffer.size() - 1;
         if (!m_current.buffer.empty() && m_current.buffer[idx] == ',') {
             m_current.buffer[idx] = '\t';
@@ -233,10 +346,16 @@ class db_copy_mgr_t
     /**
      * Add a column with the given WKB geometry in WKB hex format.
      *
-     * The geometry is converted on-the-fly from WKB binary to WKB hex.
+     * The geometry is converted on-the-fly from WKB binary to WKB hex. In
+     * the binary format the WKB is sent as is.
      */
     void add_hex_geom(std::string const &wkb)
     {
+        if (m_binary_types) {
+            check_field_type(next_field(), copy_field_type::geometry);
+            add_binary_bytes(wkb);
+            return;
+        }
         util::encode_hex(wkb, &m_current.buffer);
         m_current.buffer += '\t';
     }
@@ -277,12 +396,247 @@ class db_copy_mgr_t
     }
 
 private:
+    /// OID of the int8 type, needed for the elements of int8[].
+    static constexpr int32_t INT8OID = 20;
+    /// Array header: ndim, flags, element type, one dimension, lower bound
+    static constexpr std::size_t ARRAY_HEADER_SIZE = 20;
+    /// Array element: length and int8 value
+    static constexpr std::size_t ARRAY_ELEMENT_SIZE = 12;
+
+    /**
+     * Get the type of the next field of the current row in binary format.
+     * A row starts with the number of fields, that is written together with
+     * the first field, so that nothing is left in the buffer from a
+     * new_line() that is only used to delete objects.
+     */
+    copy_field_type next_field()
+    {
+        assert(m_binary_types);
+        assert(m_field < m_binary_types->size());
+        if (m_field == 0) {
+            put_int16(static_cast<int16_t>(m_binary_types->size()));
+        }
+        return (*m_binary_types)[m_field++];
+    }
+
+    static void check_field_type(copy_field_type type, copy_field_type expected)
+    {
+        if (type != expected) {
+            throw_type_mismatch(type);
+        }
+    }
+
+    [[noreturn]] static void throw_type_mismatch(copy_field_type type)
+    {
+        throw fmt_error("Internal error: Wrong data for column of type {} in"
+                        " binary COPY.",
+                        static_cast<int>(type));
+    }
+
+    /// Is the integer value representable in the (signed) type R?
+    template <typename R, typename T>
+    static constexpr bool fits_in(T value) noexcept
+    {
+        if constexpr (std::is_signed_v<T>) {
+            return value >= std::numeric_limits<R>::min() &&
+                   value <= std::numeric_limits<R>::max();
+        } else {
+            return static_cast<std::uintmax_t>(value) <=
+                   static_cast<std::uintmax_t>(std::numeric_limits<R>::max());
+        }
+    }
+
+    /**
+     * The bytes of an integer in big-endian (network) byte order, which is
+     * what the binary format uses. Compilers turn this into a single byte
+     * swap instruction.
+     */
+    template <typename T, std::size_t... I>
+    static constexpr std::array<char, sizeof(T)>
+    big_endian_bytes(T value, std::index_sequence<I...> /*bytes*/) noexcept
+    {
+        auto const v =
+            static_cast<uint64_t>(static_cast<std::make_unsigned_t<T>>(value));
+        return {static_cast<char>(v >> (8U * (sizeof(T) - 1 - I)))...};
+    }
+
+    template <typename T>
+    static constexpr std::array<char, sizeof(T)>
+    big_endian_bytes(T value) noexcept
+    {
+        return big_endian_bytes(value, std::make_index_sequence<sizeof(T)>{});
+    }
+
+    template <typename T>
+    void put_be(T value)
+    {
+        auto const bytes = big_endian_bytes(value);
+        m_current.buffer.append(bytes.data(), bytes.size());
+    }
+
+    void put_int16(int16_t value) { put_be(value); }
+    void put_int32(int32_t value) { put_be(value); }
+
+    /// Write the length of a field or the number of elements of something.
+    void put_length(std::size_t length)
+    {
+        assert(length <= std::numeric_limits<int32_t>::max());
+        put_int32(static_cast<int32_t>(length));
+    }
+
+    /// Overwrite an int32 written earlier at position pos.
+    void put_length_at(std::size_t pos, std::size_t length)
+    {
+        assert(length <= std::numeric_limits<int32_t>::max());
+        auto const bytes = big_endian_bytes(static_cast<int32_t>(length));
+        m_current.buffer.replace(pos, bytes.size(), bytes.data(), bytes.size());
+    }
+
+    /// Write a complete field with a fixed-size value: length, then value.
+    template <typename T>
+    void put_value(T value)
+    {
+        put_length(sizeof(T));
+        put_be(value);
+    }
+
+    /// Reserve space for the length of a field, return its position.
+    std::size_t start_field_length()
+    {
+        auto const pos = m_current.buffer.size();
+        m_current.buffer.append(4, '\0');
+        return pos;
+    }
+
+    void finish_field_length(std::size_t pos)
+    {
+        put_length_at(pos, m_current.buffer.size() - pos - 4);
+    }
+
+    void add_binary_bytes(std::string_view data)
+    {
+        put_length(data.size());
+        m_current.buffer += data;
+    }
+
+    void add_binary(copy_field_type type, std::string_view str)
+    {
+        switch (type) {
+        case copy_field_type::text:
+            add_binary_bytes(str);
+            break;
+        case copy_field_type::jsonb:
+            put_length(str.size() + 1);
+            m_current.buffer += '\1'; // jsonb format version
+            m_current.buffer += str;
+            break;
+        default:
+            throw_type_mismatch(type);
+        }
+    }
+
+    void add_binary(copy_field_type type, char const *str)
+    {
+        add_binary(type, std::string_view{str});
+    }
+
+    void add_binary(copy_field_type type, std::string const &str)
+    {
+        add_binary(type, std::string_view{str});
+    }
+
+    void add_binary(copy_field_type type, osmium::Timestamp timestamp)
+    {
+        check_field_type(type, copy_field_type::timestamptz);
+        if (!timestamp.valid()) {
+            // The text format sends an empty string, which is invalid, too.
+            throw std::runtime_error{"Invalid timestamp (0)."};
+        }
+        // Microseconds since 2000-01-01 00:00:00 UTC
+        constexpr int64_t PG_EPOCH = 946684800;
+        put_value(
+            (static_cast<int64_t>(timestamp.seconds_since_epoch()) - PG_EPOCH) *
+            1000000);
+    }
+
+    template <typename T>
+    std::enable_if_t<std::is_arithmetic_v<T>> add_binary(copy_field_type type,
+                                                         T value)
+    {
+        if constexpr (std::is_same_v<T, char>) {
+            check_field_type(type, copy_field_type::text);
+            add_binary_bytes(std::string_view{&value, 1});
+            return;
+        }
+
+        switch (type) {
+        case copy_field_type::boolean:
+            put_length(1);
+            m_current.buffer += (value != 0) ? '\1' : '\0';
+            return;
+        case copy_field_type::float4: {
+            float const f = copy_to_float4(static_cast<double>(value));
+            uint32_t bits = 0;
+            std::memcpy(&bits, &f, sizeof(bits));
+            put_value(bits);
+            return;
+        }
+        case copy_field_type::float8: {
+            auto d = static_cast<double>(value);
+            if (std::isnan(d)) {
+                d = std::numeric_limits<double>::quiet_NaN();
+            }
+            uint64_t bits = 0;
+            std::memcpy(&bits, &d, sizeof(bits));
+            put_value(bits);
+            return;
+        }
+        case copy_field_type::text:
+            add_binary_bytes(fmt::to_string(value));
+            return;
+        default:
+            break;
+        }
+
+        if constexpr (std::is_integral_v<T>) {
+            switch (type) {
+            case copy_field_type::int2:
+                assert(fits_in<int16_t>(value));
+                put_value(static_cast<int16_t>(value));
+                return;
+            case copy_field_type::int4:
+                assert(fits_in<int32_t>(value));
+                put_value(static_cast<int32_t>(value));
+                return;
+            case copy_field_type::int8:
+                put_value(static_cast<int64_t>(value));
+                return;
+            default:
+                break;
+            }
+        }
+
+        throw_type_mismatch(type);
+    }
+
+    void add_binary_hash_elem(char const *k, char const *v)
+    {
+        add_binary_bytes(k);
+        add_binary_bytes(v);
+        ++m_hash_pairs;
+    }
+
     template <typename T>
     void add_value(T value)
     {
         m_current.buffer += fmt::to_string(value);
     }
 
+    void add_value(osmium::Timestamp timestamp)
+    {
+        m_current.buffer += timestamp.to_iso();
+    }
+
     void add_value(std::string const &s) { add_value(s.c_str()); }
 
     void add_value(char const *s)
@@ -341,6 +695,15 @@ class db_copy_mgr_t
     std::shared_ptr<db_copy_thread_t> m_processor;
     db_cmd_copy_delete_t<DELETER> m_current;
     std::size_t m_committed = 0;
+
+    /// Column types of the current target if it uses the binary format.
+    std::vector<copy_field_type> const *m_binary_types = nullptr;
+    /// Number of the next field in the current row (binary format).
+    std::size_t m_field = 0;
+    /// Start of the array or hash field being written (binary format).
+    std::size_t m_field_start = 0;
+    /// Number of pairs in the hash field being written (binary format).
+    std::size_t m_hash_pairs = 0;
 };
 
 #endif // OSM2PGSQL_DB_COPY_MGR_HPP
diff --git a/src/db-copy.cpp b/src/db-copy.cpp
index e28e744a2..048aec2df 100644
--- a/src/db-copy.cpp
+++ b/src/db-copy.cpp
@@ -220,6 +220,10 @@ void db_copy_thread_t::thread_t::start_copy(
                        target->rows());
     }
 
+    if (target->binary()) {
+        fmt::format_to(std::back_inserter(sql), " (FORMAT binary)");
+    }
+
     if (!target->conditions().empty()) {
         fmt::format_to(std::back_inserter(sql), FMT_STRING(" WHERE {}"),
                        target->conditions());
@@ -228,12 +232,24 @@ void db_copy_thread_t::thread_t::start_copy(
     sql.push_back('\0');
     m_db_connection.copy_start(to_string(sql));
 
+    if (target->binary()) {
+        // Signature, flags (none), length of header extension (none)
+        static constexpr std::string_view HEADER{
+            "PGCOPY\n\xff\r\n\0\0\0\0\0\0\0\0\0", 19};
+        m_db_connection.copy_send(HEADER, target->name());
+    }
+
     m_inflight = target;
 }
 
 void db_copy_thread_t::thread_t::finish_copy()
 {
     if (m_inflight) {
+        if (m_inflight->binary()) {
+            // File trailer: a row with field count -1
+            static constexpr std::string_view TRAILER{"\xff\xff", 2};
+            m_db_connection.copy_send(TRAILER, m_inflight->name());
+        }
         m_db_connection.copy_end(m_inflight->name());
         m_inflight.reset();
     }
diff --git a/src/db-copy.hpp b/src/db-copy.hpp
index 84faf8515..e37ab216d 100644
--- a/src/db-copy.hpp
+++ b/src/db-copy.hpp
@@ -15,8 +15,9 @@
 #include "pgsql-params.hpp"
 
 #include <cassert>
-#include <cstddef>
 #include <condition_variable>
+#include <cstddef>
+#include <cstdint>
 #include <deque>
 #include <future>
 #include <memory>
@@ -27,6 +28,27 @@
 #include <variant>
 #include <vector>
 
+/**
+ * The PostgreSQL type of a column as far as the binary COPY format is
+ * concerned. Every type has its own binary representation which must match
+ * the column type exactly, the server does not convert anything.
+ */
+enum class copy_field_type : uint8_t
+{
+    text, ///< text, char(n), json: the string itself
+    boolean,
+    int2,
+    int4,
+    int8,
+    float4,
+    float8,
+    int8_array, ///< one-dimensional int8[] without NULLs
+    hstore,
+    jsonb,
+    geometry,   ///< PostGIS geometry, sent as EWKB
+    timestamptz ///< from osmium::Timestamp only
+};
+
 /**
  * Table information necessary for building SQL queries.
  */
@@ -55,6 +77,27 @@ class db_target_descr_t
         m_conditions = std::move(conditions);
     }
 
+    /**
+     * Rows are sent in PostgreSQL's binary COPY format if the types of all
+     * columns are known, in the text format otherwise.
+     */
+    bool binary() const noexcept { return !m_binary_types.empty(); }
+
+    /// The types of the columns for the binary COPY format.
+    std::vector<copy_field_type> const &binary_types() const noexcept
+    {
+        return m_binary_types;
+    }
+
+    /**
+     * Set the types of all columns (in the order of rows()) to switch to the
+     * binary COPY format. An empty vector means text format.
+     */
+    void set_binary_types(std::vector<copy_field_type> types)
+    {
+        m_binary_types = std::move(types);
+    }
+
     /**
      * Check if the buffer would use exactly the same copy operation.
      */
@@ -62,7 +105,8 @@ class db_target_descr_t
     {
         return (this == &other) ||
                (m_schema == other.m_schema && m_name == other.m_name &&
-                m_id == other.m_id && m_rows == other.m_rows);
+                m_id == other.m_id && m_rows == other.m_rows &&
+                binary() == other.binary());
     }
 
 private:
@@ -76,6 +120,8 @@ class db_target_descr_t
     std::string m_rows;
     /// Conditions for the COPY command.
     std::string m_conditions;
+    /// Column types for binary COPY format (when empty: text format).
+    std::vector<copy_field_type> m_binary_types;
 };
 
 /**
diff --git a/src/flex-table-column.cpp b/src/flex-table-column.cpp
index 98a9cd4e1..22a06231b 100644
--- a/src/flex-table-column.cpp
+++ b/src/flex-table-column.cpp
@@ -9,6 +9,7 @@
 
 #include "flex-table-column.hpp"
 
+#include "db-copy.hpp"
 #include "format.hpp"
 #include "geom-boost-adaptor.hpp"
 #include "overloaded.hpp"
@@ -128,6 +129,54 @@ void flex_table_column_t::set_projection(char const *projection)
     }
 }
 
+std::optional<copy_field_type>
+flex_table_column_t::binary_copy_type() const noexcept
+{
+    if (!m_sql_type.empty()) {
+        return {}; // we don't know the binary format of arbitrary types
+    }
+
+    switch (m_type) {
+    case table_column_type::text:
+    case table_column_type::json:
+    case table_column_type::id_type:
+        return copy_field_type::text;
+    case table_column_type::boolean:
+        return copy_field_type::boolean;
+    case table_column_type::int2:
+    case table_column_type::direction:
+        return copy_field_type::int2;
+    case table_column_type::int4:
+        return copy_field_type::int4;
+    case table_column_type::int8:
+    case table_column_type::id_num:
+        return copy_field_type::int8;
+    case table_column_type::real:
+        return copy_field_type::float4;
+    case table_column_type::double_precision:
+        return copy_field_type::float8;
+    case table_column_type::timestamp:
+    case table_column_type::timestamptz:
+        // These can get strings from Lua in any format PostgreSQL
+        // understands, so we leave the parsing to PostgreSQL.
+        return {};
+    case table_column_type::hstore:
+        return copy_field_type::hstore;
+    case table_column_type::jsonb:
+        return copy_field_type::jsonb;
+    case table_column_type::geometry:
+    case table_column_type::point:
+    case table_column_type::linestring:
+    case table_column_type::polygon:
+    case table_column_type::multipoint:
+    case table_column_type::multilinestring:
+    case table_column_type::multipolygon:
+    case table_column_type::geometrycollection:
+        return copy_field_type::geometry;
+    }
+    return {};
+}
+
 std::string flex_table_column_t::sql_type_name() const
 {
     if (!m_sql_type.empty()) {
diff --git a/src/flex-table-column.hpp b/src/flex-table-column.hpp
index 861ec7218..ec956d813 100644
--- a/src/flex-table-column.hpp
+++ b/src/flex-table-column.hpp
@@ -17,6 +17,7 @@
 
 #include <cassert>
 #include <cstdint>
+#include <optional>
 #include <string>
 #include <unordered_map>
 #include <vector>
@@ -58,6 +59,8 @@ enum class table_column_type : uint8_t
 
 class geometry_cache_t;
 
+enum class copy_field_type : uint8_t;
+
 /**
  * A column in a flex_table_t.
  */
@@ -71,6 +74,14 @@ class flex_table_column_t
 
     table_column_type type() const noexcept { return m_type; }
 
+    /**
+     * The type of this column in the binary COPY format. Nothing if the
+     * column can only be written in the text format, because it has a
+     * user-defined SQL type or because it can contain strings which only
+     * PostgreSQL can parse (timestamps).
+     */
+    std::optional<copy_field_type> binary_copy_type() const noexcept;
+
     bool is_point_column() const noexcept
     {
         return (m_type == table_column_type::point) ||
diff --git a/src/flex-table.cpp b/src/flex-table.cpp
index fbd16f704..c83c349c3 100644
--- a/src/flex-table.cpp
+++ b/src/flex-table.cpp
@@ -207,6 +207,22 @@ flex_table_t::build_sql_create_table(table_type ttype,
     return sql;
 }
 
+std::vector<copy_field_type> flex_table_t::binary_copy_types() const
+{
+    std::vector<copy_field_type> types;
+    for (auto const &column : m_columns) {
+        if (column.create_only()) {
+            continue;
+        }
+        auto const type = column.binary_copy_type();
+        if (!type) {
+            return {};
+        }
+        types.push_back(*type);
+    }
+    return types;
+}
+
 std::string flex_table_t::build_sql_column_list() const
 {
     assert(!m_columns.empty());
@@ -295,6 +311,19 @@ namespace {
 void table_connection_t::start(pg_conn_t const &db_connection,
                                bool append) const
 {
+    if (m_target->binary()) {
+        log_debug("Table '{}' uses the binary COPY format.", table().name());
+    } else {
+        for (auto const &column : table().columns()) {
+            if (!column.create_only() && !column.binary_copy_type()) {
+                log_debug("Table '{}' uses the text COPY format because of"
+                          " column '{}'.",
+                          table().name(), column.name());
+                break;
+            }
+        }
+    }
+
     if (!append) {
         drop_table_if_exists(db_connection, table().schema(), table().name());
     }
diff --git a/src/flex-table.hpp b/src/flex-table.hpp
index be932f6c7..6c7d41f8e 100644
--- a/src/flex-table.hpp
+++ b/src/flex-table.hpp
@@ -159,6 +159,12 @@ class flex_table_t
 
     std::string build_sql_copy_condition() const;
 
+    /**
+     * The types of all columns written by COPY for the binary COPY format.
+     * Empty if the table must use the text format.
+     */
+    std::vector<copy_field_type> binary_copy_types() const;
+
     std::string build_sql_create_id_index() const;
 
     /// Does this table take objects of the specified type?
@@ -293,6 +299,7 @@ class table_connection_t
           table->build_sql_column_list(), table->build_sql_copy_condition())),
       m_copy_mgr(copy_thread)
     {
+        m_target->set_binary_types(table->binary_copy_types());
     }
 
     void start(pg_conn_t const &db_connection, bool append) const;
diff --git a/src/middle-pgsql.cpp b/src/middle-pgsql.cpp
index 454d2b146..3afcde4af 100644
--- a/src/middle-pgsql.cpp
+++ b/src/middle-pgsql.cpp
@@ -18,12 +18,14 @@
 #include <cassert>
 #include <cstdint>
 #include <cstdlib>
+#include <initializer_list>
 #include <iterator>
 #include <memory>
 #include <stdexcept>
 #include <string>
 #include <unordered_map>
 #include <utility>
+#include <vector>
 
 #include <osmium/builder/osm_object_builder.hpp>
 #include <osmium/memory/buffer.hpp>
@@ -356,7 +358,7 @@ void members_to_json(osmium::RelationMemberList const &members,
 void middle_pgsql_t::copy_attributes(osmium::OSMObject const &obj)
 {
     if (obj.timestamp()) {
-        m_db_copy.add_column(obj.timestamp().to_iso());
+        m_db_copy.add_column(obj.timestamp());
     } else {
         m_db_copy.add_null_column();
     }
@@ -1077,6 +1079,8 @@ void middle_pgsql_t::write_users_table()
 
     auto const users_table = std::make_shared<db_target_descr_t>(
         m_options->dbschema, table_name, "id");
+    users_table->set_binary_types(
+        {copy_field_type::int4, copy_field_type::text});
 
     for (auto const &[id, name] : m_users) {
         m_db_copy.new_line(users_table);
@@ -1266,6 +1270,26 @@ middle_pgsql_t::middle_pgsql_t(std::shared_ptr<thread_pool_t> thread_pool,
     m_tables.nodes() = table_desc_t{*options, "nodes"};
     m_tables.ways() = table_desc_t{*options, "ways"};
     m_tables.relations() = table_desc_t{*options, "rels"};
+
+    // We know the types of all columns of the middle tables (see
+    // table_setup()), so they always use the binary COPY format.
+    using cft = copy_field_type;
+    std::vector<cft> attributes;
+    if (m_store_options.with_attributes) {
+        attributes = {cft::timestamptz, cft::int4, cft::int4, cft::int4};
+    }
+    auto const types = [&](std::vector<cft> columns,
+                           std::initializer_list<cft> after_attributes) {
+        columns.insert(columns.end(), attributes.cbegin(), attributes.cend());
+        columns.insert(columns.end(), after_attributes);
+        return columns;
+    };
+    m_tables.nodes().copy_target()->set_binary_types(
+        types({cft::int8, cft::int4, cft::int4}, {cft::jsonb}));
+    m_tables.ways().copy_target()->set_binary_types(
+        types({cft::int8}, {cft::int8_array, cft::jsonb}));
+    m_tables.relations().copy_target()->set_binary_types(
+        types({cft::int8}, {cft::jsonb, cft::jsonb}));
 }
 
 void middle_pgsql_t::set_requirements(
diff --git a/tests/test-db-copy-mgr.cpp b/tests/test-db-copy-mgr.cpp
index 00ec098b5..ff023855a 100644
--- a/tests/test-db-copy-mgr.cpp
+++ b/tests/test-db-copy-mgr.cpp
@@ -9,12 +9,16 @@
 
 #include <catch.hpp>
 
+#include <cmath>
+#include <limits>
 #include <string>
 #include <utility>
 #include <vector>
 
 #include "common-pg.hpp"
 #include "db-copy-mgr.hpp"
+#include "geom.hpp"
+#include "wkb.hpp"
 
 namespace {
 
@@ -88,6 +92,67 @@ void check_row(std::vector<std::string> const &row)
     }
 }
 
+/**
+ * Write a row with the value added by the write function into a table with
+ * the text format and into one with the binary format, then check that the
+ * database ends up with the same value in both.
+ */
+template <typename FUNC>
+void check_text_and_binary(std::string const &sql_type, copy_field_type type,
+                           FUNC const &write)
+{
+    auto const conn = db.connect();
+    conn.exec("DROP TABLE IF EXISTS test_copy_text, test_copy_binary");
+    conn.exec("CREATE TABLE test_copy_text (id int8, v {})", sql_type);
+    conn.exec("CREATE TABLE test_copy_binary (id int8, v {})", sql_type);
+
+    auto const text =
+        std::make_shared<db_target_descr_t>("public", "test_copy_text", "id");
+    auto const binary =
+        std::make_shared<db_target_descr_t>("public", "test_copy_binary", "id");
+    binary->set_binary_types({copy_field_type::int8, type});
+
+    copy_mgr_t mgr{std::make_shared<db_copy_thread_t>(db.connection_params())};
+    for (auto const &target : {text, binary}) {
+        mgr.new_line(target);
+        mgr.add_column(1);
+        write(&mgr);
+        mgr.finish_line();
+    }
+    mgr.sync();
+
+    auto const res =
+        conn.exec("SELECT t.v::text, b.v::text, t.v IS NULL, b.v IS NULL"
+                  " FROM test_copy_text t, test_copy_binary b");
+    REQUIRE(res.num_tuples() == 1);
+    CHECK(std::string{res.get_value(0, 0)} == res.get_value(0, 1));
+    CHECK(std::string{res.get_value(0, 2)} == res.get_value(0, 3));
+}
+
+template <typename T>
+void check_value(std::string const &sql_type, copy_field_type type, T value)
+{
+    check_text_and_binary(sql_type, type,
+                          [&](copy_mgr_t *mgr) { mgr->add_column(value); });
+    check_text_and_binary(sql_type, type,
+                          [&](copy_mgr_t *mgr) { mgr->add_null_column(); });
+}
+
+/// Write a single value in binary format, the write is expected to throw.
+template <typename T>
+void check_binary_throws(copy_field_type type, T value)
+{
+    auto const t = setup_table("v text");
+    t->set_binary_types({copy_field_type::int8, type});
+
+    copy_mgr_t mgr{std::make_shared<db_copy_thread_t>(db.connection_params())};
+    mgr.new_line(t);
+    mgr.add_column(1);
+    REQUIRE_THROWS(mgr.add_column(value));
+    mgr.rollback_line();
+    mgr.sync();
+}
+
 } // anonymous namespace
 
 TEST_CASE("copy_mgr_t: Insert null")
@@ -229,3 +294,193 @@ TEST_CASE("copy_mgr_t: Insert something, insert more, roll back, insert "
     CHECK(res.get(0, 0) == "good");
     CHECK(res.get(1, 0) == "better");
 }
+
+TEST_CASE("copy_mgr_t: Binary format gives the same text values as text")
+{
+    for (char const *str : {"foo", "", "va\tr\nK\\P\r\"quoted\"",
+                            "\xc3\xa4\xe6\xbc\xa2\xf0\x9f\x99\x82"}) {
+        check_value("text", copy_field_type::text, str);
+        check_value("text", copy_field_type::text, std::string{str});
+    }
+    check_value("char(1)", copy_field_type::text, 'N');
+    check_value("text", copy_field_type::text, 42);
+    check_value("json", copy_field_type::text, R"({"a": [1, 2.5]})");
+}
+
+TEST_CASE("copy_mgr_t: Binary format gives the same booleans as text")
+{
+    check_value("boolean", copy_field_type::boolean, true);
+    check_value("boolean", copy_field_type::boolean, false);
+    check_value("boolean", copy_field_type::boolean, 1);
+}
+
+TEST_CASE("copy_mgr_t: Binary format gives the same integers as text")
+{
+    for (int64_t const v :
+         {int64_t{0}, int64_t{-1}, int64_t{-32768}, int64_t{32767}}) {
+        check_value("int2", copy_field_type::int2, v);
+    }
+    for (int64_t const v : {int64_t{std::numeric_limits<int32_t>::min()},
+                            int64_t{std::numeric_limits<int32_t>::max()}}) {
+        check_value("int4", copy_field_type::int4, v);
+    }
+    for (int64_t const v : {std::numeric_limits<int64_t>::min(),
+                            std::numeric_limits<int64_t>::max()}) {
+        check_value("int8", copy_field_type::int8, v);
+    }
+    check_value("int4", copy_field_type::int4, -1);
+}
+
+TEST_CASE("copy_mgr_t: Binary format gives the same reals as text")
+{
+    for (double const v :
+         {0.0, -0.0, 0.1, 1.5, -123456.789, 3.4028235e38, 1e-40, 1e-45,
+          std::numeric_limits<double>::quiet_NaN(),
+          -std::numeric_limits<double>::quiet_NaN(),
+          std::numeric_limits<double>::infinity(),
+          -std::numeric_limits<double>::infinity()}) {
+        check_value("real", copy_field_type::float4, v);
+        check_value("double precision", copy_field_type::float8, v);
+    }
+    check_value("double precision", copy_field_type::float8, 1e300);
+    check_value("real", copy_field_type::float4, 7);
+}
+
+TEST_CASE("copy_mgr_t: Binary format rejects reals out of range like text")
+{
+    check_binary_throws(copy_field_type::float4, 1e39);
+    check_binary_throws(copy_field_type::float4, -1e39);
+    check_binary_throws(copy_field_type::float4, 1e-50);
+}
+
+TEST_CASE("copy_mgr_t: Binary format gives the same jsonb as text")
+{
+    check_value("jsonb", copy_field_type::jsonb,
+                R"({"b": null, "a": [1, 2.5, "x\ty"], "c": "ä"})");
+    check_value("jsonb", copy_field_type::jsonb, std::string{"[]"});
+}
+
+TEST_CASE("copy_mgr_t: Binary format rounds doubles to the nearest float")
+{
+    // The text format rounds twice (to the shortest decimal representation,
+    // then to float) which gives a different result for values exactly
+    // halfway between two floats. The binary format rounds once, correctly.
+    auto const t = setup_table("v real");
+    t->set_binary_types({copy_field_type::int8, copy_field_type::float4});
+
+    copy_mgr_t mgr{std::make_shared<db_copy_thread_t>(db.connection_params())};
+    double const halfway_down = 1.0 + std::ldexp(1.0, -24);   // to even: 1
+    double const halfway_up = 1.0 + 3 * std::ldexp(1.0, -24); // 1 + 2^-22
+    add_row(&mgr, t, 1, halfway_down);
+    add_row(&mgr, t, 2, halfway_up);
+
+    auto const conn = db.connect();
+    CHECK(conn.result_as_int("SELECT count(*) FROM test_copy_mgr WHERE"
+                             " (id = 1 AND v = 1::real) OR"
+                             " (id = 2 AND v = (1 + 2 ^ (-22))::real)") == 2);
+}
+
+TEST_CASE("copy_mgr_t: Binary format gives the same timestamps as text")
+{
+    for (uint32_t const seconds : {1U, 946684800U, 1234567890U, 0xffffffffU}) {
+        check_value("timestamptz", copy_field_type::timestamptz,
+                    osmium::Timestamp{seconds});
+    }
+    check_binary_throws(copy_field_type::timestamptz, osmium::Timestamp{});
+}
+
+TEST_CASE("copy_mgr_t: Binary format gives the same arrays as text")
+{
+    for (std::vector<osmid_t> const &values :
+         {std::vector<osmid_t>{}, std::vector<osmid_t>{1, -2, 3},
+          std::vector<osmid_t>{std::numeric_limits<osmid_t>::max()}}) {
+        check_text_and_binary("int8[]", copy_field_type::int8_array,
+                              [&](copy_mgr_t *mgr) {
+                                  mgr->new_array();
+                                  for (auto const v : values) {
+                                      mgr->add_array_elem(v);
+                                  }
+                                  mgr->finish_array();
+                              });
+    }
+}
+
+TEST_CASE("copy_mgr_t: Binary format gives the same hstore as text")
+{
+    std::vector<std::pair<std::string, std::string>> const values = {
+        {"one", "two"},
+        {"key 1", "value 1"},
+        {"\"key\"", "\"value\""},
+        {"key\t2", "value\t2"},
+        {"key\n3", "value\n3"},
+        {"key\\5", "value\\5"},
+        {"", ""}};
+
+    check_text_and_binary("hstore", copy_field_type::hstore,
+                          [&](copy_mgr_t *mgr) {
+                              mgr->new_hash();
+                              for (auto const &[k, v] : values) {
+                                  mgr->add_hash_elem(k, v);
+                              }
+                              mgr->add_hstore_num_noescape("num", 17);
+                              mgr->finish_hash();
+                          });
+
+    check_text_and_binary("hstore", copy_field_type::hstore,
+                          [&](copy_mgr_t *mgr) {
+                              mgr->new_hash();
+                              mgr->finish_hash();
+                          });
+}
+
+TEST_CASE("copy_mgr_t: Binary format gives the same geometries as text")
+{
+    geom::geometry_t point{geom::point_t{1.5, -2.25}};
+    point.set_srid(3857);
+    geom::geometry_t line{geom::linestring_t{{0, 0}, {1, 1}, {2, 0}}};
+    line.set_srid(4326);
+
+    for (auto const *geom : {&point, &line}) {
+        for (bool const wrap_multi : {false, true}) {
+            auto const wkb = geom_to_ewkb(*geom, wrap_multi);
+            check_text_and_binary(
+                "geometry", copy_field_type::geometry,
+                [&](copy_mgr_t *mgr) { mgr->add_hex_geom(wkb); });
+        }
+    }
+}
+
+TEST_CASE("copy_mgr_t: Binary format with deletes, rollback and many rows")
+{
+    auto const t = setup_table("t text, n int4");
+    t->set_binary_types(
+        {copy_field_type::int8, copy_field_type::text, copy_field_type::int4});
+
+    copy_mgr_t mgr{std::make_shared<db_copy_thread_t>(db.connection_params())};
+
+    // Long enough so that the rows need several buffers
+    std::string const text(200, 'x');
+
+    // A new_line() just for deleting must not leave anything in the buffer.
+    mgr.new_line(t);
+    mgr.delete_object(1);
+
+    for (int i = 0; i < 100000; ++i) {
+        mgr.new_line(t);
+        mgr.add_column(i);
+        mgr.add_column(text);
+        if (i % 3 == 0) {
+            mgr.rollback_line();
+            continue;
+        }
+        mgr.add_column(i * 2);
+        mgr.finish_line();
+    }
+    mgr.sync();
+
+    auto const conn = db.connect();
+    CHECK(conn.get_count("test_copy_mgr") == 66666);
+    CHECK(conn.result_as_int("SELECT count(*) FROM test_copy_mgr"
+                             " WHERE n = id * 2 AND t = repeat('x', 200)") ==
+          66666);
+}