split middle-pgsql into insert and query class

The query instance can work on a single connection. This saves
a couple of connections in the processing stage.

Requires slightly changing the behviour of the connections
that copy in the data. They need to commit, so that the
data becomes visible.

middle also needs to start the tables earlier, because
the tables must have been created when preparing queries
in the query connection.

Author: lonvia

middle-pgsql.cpp

@@ -17,29 +17,54 @@
 #include "node-ram-cache.hpp"
 #include "pgsql.hpp"
 
-struct middle_pgsql_t : public slim_middle_t, public middle_query_t
+class middle_query_pgsql_t : public middle_query_t
 {
-    middle_pgsql_t();
+public:
+    middle_query_pgsql_t(
+        char const *conninfo, std::shared_ptr<node_ram_cache> const &cache,
+        std::shared_ptr<node_persistent_cache> const &persistent_cache);
+    ~middle_query_pgsql_t();
 
-    void start(const options_t *out_options_) override;
+    size_t nodes_get_list(osmium::WayNodeList *nodes) const override;
+
+    bool ways_get(osmid_t id, osmium::memory::Buffer &buffer) const override;
+    size_t rel_way_members_get(osmium::Relation const &rel, rolelist_t *roles,
+                               osmium::memory::Buffer &buffer) const override;
+
+    idlist_t relations_using_way(osmid_t way_id) const override;
+    bool relations_get(osmid_t id,
+                       osmium::memory::Buffer &buffer) const override;
+
+    void exec_sql(std::string const &sql_cmd) const;
+
+private:
+    size_t local_nodes_get_list(osmium::WayNodeList *nodes) const;
+
+    pg_result_t exec_prepared(char const *stmt, char const *param) const;
+    pg_result_t exec_prepared(char const *stmt, osmid_t osm_id) const;
+
+    struct pg_conn *m_sql_conn;
+    std::shared_ptr<node_ram_cache> m_cache;
+    std::shared_ptr<node_persistent_cache> m_persistent_cache;
+};
+
+struct middle_pgsql_t : public slim_middle_t
+{
+    middle_pgsql_t(options_t const *options);
+
+    void start() override;
     void stop(osmium::thread::Pool &pool) override;
     void analyze() override;
     void commit() override;
 
     void nodes_set(osmium::Node const &node) override;
-    size_t nodes_get_list(osmium::WayNodeList *nodes) const override;
     void nodes_delete(osmid_t id) override;
     void node_changed(osmid_t id) override;
 
     void ways_set(osmium::Way const &way) override;
-    bool ways_get(osmid_t id, osmium::memory::Buffer &buffer) const override;
-    size_t rel_way_members_get(osmium::Relation const &rel, rolelist_t *roles,
-                               osmium::memory::Buffer &buffer) const override;
-
     void ways_delete(osmid_t id) override;
     void way_changed(osmid_t id) override;
 
-    bool relations_get(osmid_t id, osmium::memory::Buffer &buffer) const override;
     void relations_set(osmium::Relation const &rel) override;
     void relations_delete(osmid_t id) override;
     void relation_changed(osmid_t id) override;
@@ -51,13 +76,12 @@ struct middle_pgsql_t : public slim_middle_t, public middle_query_t
 
     size_t pending_count() const override;
 
-    idlist_t relations_using_way(osmid_t way_id) const override;
-
     class table_desc
     {
     public:
         table_desc() : sql_conn(nullptr) {}
-        table_desc(char const *name, char const *create,
+        table_desc(options_t const *options, char const *name,
+                   char const *create, char const *prepare_query,
                    char const *prepare = "", char const *prepare_intarray = "",
                    char const *array_indexes = "");
 
@@ -68,8 +92,9 @@ struct middle_pgsql_t : public slim_middle_t, public middle_query_t
 
         int copyMode;    /* True if we are in copy mode */
         struct pg_conn *sql_conn;
+        std::string m_prepare_query;
 
-        void connect(options_t const *options);
+        void connect(char const *conninfo);
         void begin();
         void prepare_queries(bool append);
         void create();
@@ -109,7 +134,6 @@ struct middle_pgsql_t : public slim_middle_t, public middle_query_t
      * Sets up sql_conn for the table
      */
     void local_nodes_set(osmium::Node const &node);
-    size_t local_nodes_get_list(osmium::WayNodeList *nodes) const;
     void local_nodes_delete(osmid_t osm_id);
 
     table_desc tables[NUM_TABLES];

middle-pgsql.hpp

@@ -155,11 +155,7 @@ void middle_ram_t::analyze(void)
     /* No need */
 }
 
-void middle_ram_t::start(const options_t *options)
-{
-    extra_attributes = options->extra_attributes;
-    cache.reset(new node_ram_cache(options->alloc_chunkwise, options->cache));
-}
+void middle_ram_t::start() {}
 
 void middle_ram_t::stop(osmium::thread::Pool &pool)
 {
@@ -172,8 +168,10 @@ void middle_ram_t::stop(osmium::thread::Pool &pool)
 void middle_ram_t::commit(void) {
 }
 
-middle_ram_t::middle_ram_t():
-    ways(), rels(), cache(), simulate_ways_deleted(false)
+middle_ram_t::middle_ram_t(options_t const *options)
+: ways(), rels(),
+  cache(new node_ram_cache(options->alloc_chunkwise, options->cache)),
+  extra_attributes(options->extra_attributes), simulate_ways_deleted(false)
 {
 }
 

middle-ram.cpp

@@ -82,10 +82,10 @@ class elem_cache_t
 
 struct middle_ram_t : public middle_t, public middle_query_t
 {
-    middle_ram_t();
+    middle_ram_t(options_t const *options);
     virtual ~middle_ram_t();
 
-    void start(const options_t *options) override;
+    void start() override;
     void stop(osmium::thread::Pool &pool) override;
     void analyze(void) override;
     void commit(void) override;

middle-ram.hpp

@@ -17,8 +17,6 @@
 #include "osmtypes.hpp"
 #include "reprojection.hpp"
 
-struct options_t;
-
 /**
  * Infterface for returning information about raw OSM input data from a cache.
  */
@@ -84,7 +82,7 @@ struct middle_t
 {
     virtual ~middle_t() = 0;
 
-    virtual void start(options_t const *out_options) = 0;
+    virtual void start() = 0;
     virtual void stop(osmium::thread::Pool &pool) = 0;
     virtual void analyze(void) = 0;
     virtual void commit(void) = 0;

middle.hpp

@@ -56,20 +56,17 @@ int main(int argc, char *argv[])
 
         //setup the middle and backend (output)
         std::shared_ptr<middle_t> middle;
-        std::vector<std::shared_ptr<output_t>> outputs;
 
         if (options.slim) {
-            auto mid = std::make_shared<middle_pgsql_t>();
-            outputs = output_t::create_outputs(
-                std::static_pointer_cast<middle_query_t>(mid), options);
-            middle = std::static_pointer_cast<middle_t>(mid);
+            middle = std::shared_ptr<middle_t>(new middle_pgsql_t(&options));
         } else {
-            auto mid = std::make_shared<middle_ram_t>();
-            outputs = output_t::create_outputs(
-                std::static_pointer_cast<middle_query_t>(mid), options);
-            middle = std::static_pointer_cast<middle_t>(mid);
+            middle = std::shared_ptr<middle_t>(new middle_ram_t(&options));
         }
 
+        middle->start();
+
+        auto outputs = output_t::create_outputs(
+            middle->get_query_instance(middle), options);
         //let osmdata orchestrate between the middle and the outs
         osmdata_t osmdata(middle, outputs);
 

osm2pgsql.cpp

@@ -171,7 +171,6 @@ void osmdata_t::start() {
     for (auto& out: outs) {
         out->start();
     }
-    mid->start(outs[0]->get_options());
 }
 
 void osmdata_t::type_changed(osmium::item_type new_type)

osmdata.cpp

@@ -22,14 +22,15 @@ namespace testing {
                        char const *file_format)
     {
         //setup the middle
-        auto middle = std::make_shared<MID>();
+        std::shared_ptr<middle_t> middle(new MID(&options));
+        middle->start();
 
         //setup the backend (output)
         auto outputs = output_t::create_outputs(
-            std::static_pointer_cast<middle_query_t>(middle), options);
+            middle->get_query_instance(middle), options);
 
         //let osmdata orchestrate between the middle and the outs
-        osmdata_t osmdata(std::static_pointer_cast<middle_t>(middle), outputs);
+        osmdata_t osmdata(middle, outputs);
 
         parse(test_file, file_format, options, &osmdata);
     }

tests/common.hpp

@@ -36,24 +36,24 @@ class test_middle_helper
 
 public:
     test_middle_helper(options_t const &options)
-    : m_mid(std::make_shared<MID>()), m_output(m_mid, options),
+    : m_mid(std::make_shared<MID>(&options)),
       m_buffer(4096, osmium::memory::Buffer::auto_grow::yes),
       m_proj(reprojection::create_projection(PROJ_LATLONG))
     {
-        m_mid->start(&options);
+        m_mid->start();
+        m_mid_q = m_mid->get_query_instance(m_mid);
+        m_output.reset(new output_null_t(m_mid_q, options));
     }
 
-    ~test_middle_helper() { commit_and_stop(); }
-
-    void start(options_t const *options) { m_mid->start(options); }
-
-    void commit_and_stop()
+    ~test_middle_helper()
     {
         osmium::thread::Pool pool(1);
         m_mid->commit();
         m_mid->stop(pool);
     }
 
+    void commit() { m_mid->commit(); }
+
     // tests that a single node can be set and retrieved. returns 0 on success.
     int test_node_set()
     {
@@ -65,9 +65,10 @@ class test_middle_helper
 
         // set the node
         m_mid->nodes_set(node);
+        m_mid->flush(osmium::item_type::way);
 
         // get it back
-        if (m_mid->nodes_get_list(&(way.nodes())) != way.nodes().size()) {
+        if (m_mid_q->nodes_get_list(&(way.nodes())) != way.nodes().size()) {
             std::cerr << "ERROR: Unable to get node list.\n";
             return 1;
         }
@@ -128,9 +129,11 @@ class test_middle_helper
             ids.push_back(node.id());
         }
 
+        m_mid->flush(osmium::item_type::way);
+
         auto &way = m_buffer.get<osmium::Way>(way_with_nodes(ids));
 
-        if (m_mid->nodes_get_list(&(way.nodes())) != ids.size()) {
+        if (m_mid_q->nodes_get_list(&(way.nodes())) != ids.size()) {
             std::cerr << "ERROR: Unable to get node list.\n";
             return 1;
         }
@@ -191,7 +194,7 @@ class test_middle_helper
 
         auto buf_pos = m_buffer.committed();
         rolelist_t roles;
-        size_t way_count = m_mid->rel_way_members_get(rel, &roles, m_buffer);
+        size_t way_count = m_mid_q->rel_way_members_get(rel, &roles, m_buffer);
         if (way_count != 1) {
             std::cerr << "ERROR: Unable to get way list.\n";
             return 1;
@@ -222,7 +225,7 @@ class test_middle_helper
                       << ", but got back " << way.id() << " from middle.\n";
             return 1;
         }
-        m_mid->nodes_get_list(&(way.nodes()));
+        m_mid_q->nodes_get_list(&(way.nodes()));
         for (size_t i = 0; i < nds.size(); ++i) {
             if (way.nodes()[i].location().lon() != lon) {
                 std::cerr << "ERROR: Way node should have lon=" << lon
@@ -309,7 +312,8 @@ class test_middle_helper
     static constexpr double test_lat(osmid_t id) { return 1 + 1e-5 * id; }
 
     std::shared_ptr<MID> m_mid;
-    output_null_t m_output;
+    std::shared_ptr<middle_query_t> m_mid_q;
+    std::unique_ptr<output_null_t> m_output;
 
     // simple osmium buffer to store all the objects in
     osmium::memory::Buffer m_buffer;

tests/middle-tests.hpp

@@ -7,7 +7,7 @@
 struct dummy_middle_t : public middle_t {
     virtual ~dummy_middle_t() = default;
 
-    void start(const options_t *) override { }
+    void start() override {}
     void stop(osmium::thread::Pool &) override {}
     void flush(osmium::item_type) override {}
     void cleanup(void) { }
@@ -26,7 +26,7 @@ struct dummy_middle_t : public middle_t {
     virtual size_t pending_count() const override  { return 0; }
 
     std::shared_ptr<middle_query_t>
-    get_query_instance(std::shared_ptr<middle_t> const &mid) const override
+    get_query_instance(std::shared_ptr<middle_t> const &) const override
     {
         return std::shared_ptr<middle_query_t>();
     }
@@ -35,7 +35,7 @@ struct dummy_middle_t : public middle_t {
 struct dummy_slim_middle_t : public slim_middle_t {
     virtual ~dummy_slim_middle_t() = default;
 
-    void start(const options_t *) override  { }
+    void start() override {}
     void stop(osmium::thread::Pool &) override {}
     void flush(osmium::item_type) override {}
     void cleanup(void) { }
@@ -54,7 +54,7 @@ struct dummy_slim_middle_t : public slim_middle_t {
     size_t pending_count() const override  { return 0; }
 
     std::shared_ptr<middle_query_t>
-    get_query_instance(std::shared_ptr<middle_t> const &mid) const override
+    get_query_instance(std::shared_ptr<middle_t> const &) const override
     {
         return std::shared_ptr<middle_query_t>();
     }