Merge pull request #1904 from joto/flush-copy

Flush COPY between nodes/ways and ways/relations respectively

Author: lonvia

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

@@ -1001,6 +1001,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()