Flush COPY between nodes/ways and ways/relations respectively

In most cases database tables are filled based on OSM object types, so
after we have read all nodes nothing will be added to node-type tables
any more etc. So it makes sense to explicitly flush the COPY making sure
this data is written to the database. If more data is later added, maybe
because of second-stage processing, it doesn't matter, the COPY will be
reopened.

Author: joto

src/db-copy-mgr.hpp

@@ -271,18 +271,22 @@ class db_copy_mgr_t
         m_current->add_deletable(std::forward<ARGS>(args)...);
     }
 
+    void flush()
+    {
+        // finish any ongoing copy operations
+        if (m_current) {
+            m_processor->add_buffer(std::move(m_current));
+        }
+    }
+
     /**
      * Synchronize with worker.
      *
      * Only returns when all previously issued commands are done.
      */
     void sync()
     {
-        // finish any ongoing copy operations
-        if (m_current) {
-            m_processor->add_buffer(std::move(m_current));
-        }
-
+        flush();
         m_processor->sync_and_wait();
     }
 

src/flex-table.hpp

@@ -247,6 +247,8 @@ class table_connection_t
 
     pg_result_t get_geom_by_id(osmium::item_type type, osmid_t id) const;
 
+    void flush() { m_copy_mgr.flush(); }
+
     void sync() { m_copy_mgr.sync(); }
 
     void new_line() { m_copy_mgr.new_line(m_target); }

src/osmdata.cpp

@@ -67,7 +67,11 @@ void osmdata_t::node(osmium::Node const &node)
     }
 }
 
-void osmdata_t::after_nodes() { m_mid->after_nodes(); }
+void osmdata_t::after_nodes()
+{
+    m_mid->after_nodes();
+    m_output->after_nodes();
+}
 
 void osmdata_t::way(osmium::Way &way)
 {
@@ -84,7 +88,11 @@ void osmdata_t::way(osmium::Way &way)
     }
 }
 
-void osmdata_t::after_ways() { m_mid->after_ways(); }
+void osmdata_t::after_ways()
+{
+    m_mid->after_ways();
+    m_output->after_ways();
+}
 
 void osmdata_t::relation(osmium::Relation const &rel)
 {

src/output-flex.cpp

@@ -1263,6 +1263,20 @@ void output_flex_t::sync()
     }
 }
 
+void output_flex_t::after_nodes()
+{
+    for (auto &table : m_table_connections) {
+        table.flush();
+    }
+}
+
+void output_flex_t::after_ways()
+{
+    for (auto &table : m_table_connections) {
+        table.flush();
+    }
+}
+
 void output_flex_t::stop()
 {
     for (auto &table : m_table_connections) {

src/output-flex.hpp

@@ -124,6 +124,9 @@ class output_flex_t : public output_t
     void stop() override;
     void sync() override;
 
+    void after_nodes() override;
+    void after_ways() override;
+
     void wait() override;
 
     idset_t const &get_marked_way_ids() override;

src/output.hpp

@@ -62,6 +62,9 @@ class output_t
     virtual void stop() = 0;
     virtual void sync() = 0;
 
+    virtual void after_nodes() {}
+    virtual void after_ways() {}
+
     virtual void wait() {}
 
     virtual osmium::index::IdSetSmall<osmid_t> const &get_marked_way_ids()