Merge pull request #899 from lonvia/split-pgsql-middle

Split pgsql middle into query and insertion class

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>();
     }