Merge pull request #1371 from joto/refactor-middle

Refactor middle

Author: lonvia

src/CMakeLists.txt

@@ -8,6 +8,7 @@ set(osm2pgsql_lib_SOURCES
   geometry-processor.cpp
   input.cpp
   logging.cpp
+  middle.cpp
   middle-pgsql.cpp
   middle-ram.cpp
   node-persistent-cache.cpp

src/flex-table.cpp

@@ -256,7 +256,7 @@ void table_connection_t::stop(bool updateable, bool append)
     }
 
     log_info("Analyzing table '{}'...", table().name());
-    m_db_connection->exec("ANALYZE " + table().full_name());
+    analyze_table(*m_db_connection, table().schema(), table().name());
 
     log_info("All postprocessing on table '{}' done in {}.", table().name(),
              util::human_readable_duration(timer.stop()));

src/input.cpp

@@ -204,24 +204,59 @@ prepare_input_files(std::vector<std::string> const &input_files,
     return files;
 }
 
-static void apply(osmium::OSMObject &object, osmdata_t &osmdata,
-                  progress_display_t &progress)
+class input_context_t
 {
-    static osmium::item_type last_type = osmium::item_type::node;
-
-    if (last_type != object.type()) {
-        if (last_type == osmium::item_type::node) {
-            osmdata.after_nodes();
-            progress.start_way_counter();
-        } else if (last_type == osmium::item_type::way) {
-            osmdata.after_ways();
-            progress.start_relation_counter();
+public:
+    input_context_t(osmdata_t &osmdata, progress_display_t &progress,
+                    bool append)
+    : m_osmdata(osmdata), m_progress(progress), m_append(append)
+    {}
+
+    void apply(osmium::OSMObject &object)
+    {
+        if (!m_append && object.deleted()) {
+            throw std::runtime_error{"Input file contains deleted objects but "
+                                     "you are not in append mode."};
         }
-        last_type = object.type();
+
+        if (m_last_type != object.type()) {
+            if (m_last_type == osmium::item_type::node) {
+                m_osmdata.after_nodes();
+                m_progress.start_way_counter();
+            }
+            if (object.type() == osmium::item_type::relation) {
+                m_osmdata.after_ways();
+                m_progress.start_relation_counter();
+            }
+            m_last_type = object.type();
+        }
+
+        osmium::apply_item(object, m_osmdata, m_progress);
     }
 
-    osmium::apply_item(object, osmdata, progress);
-}
+    void eof()
+    {
+        switch (m_last_type) {
+        case osmium::item_type::node:
+            m_osmdata.after_nodes();
+            // fallthrough
+        case osmium::item_type::way:
+            m_osmdata.after_ways();
+            break;
+        default:
+            break;
+        }
+
+        m_osmdata.after_relations();
+        m_progress.print_summary();
+    }
+
+private:
+    osmdata_t &m_osmdata;
+    progress_display_t &m_progress;
+    osmium::item_type m_last_type = osmium::item_type::node;
+    bool m_append;
+}; // class input_context_t
 
 static void process_single_file(osmium::io::File const &file,
                                 osmdata_t &osmdata,
@@ -230,17 +265,14 @@ static void process_single_file(osmium::io::File const &file,
     osmium::io::Reader reader{file};
     type_id_version last{osmium::item_type::node, 0, 0};
 
+    input_context_t ctx{osmdata, progress, append};
     while (osmium::memory::Buffer buffer = reader.read()) {
         for (auto &object : buffer.select<osmium::OSMObject>()) {
             last = check_input(last, object);
-            if (!append && object.deleted()) {
-                throw std::runtime_error{
-                    "Input file contains deleted objects but "
-                    "you are not in append mode."};
-            }
-            apply(object, osmdata, progress);
+            ctx.apply(object);
         }
     }
+    ctx.eof();
 
     reader.close();
 }
@@ -262,23 +294,20 @@ static void process_multiple_files(std::vector<osmium::io::File> const &files,
         }
     }
 
+    input_context_t ctx{osmdata, progress, append};
     while (!queue.empty()) {
         auto element = queue.top();
         queue.pop();
         if (queue.empty() || element != queue.top()) {
-            if (!append && element.object().deleted()) {
-                throw std::runtime_error{
-                    "Input file contains deleted objects but "
-                    "you are not in append mode."};
-            }
-            apply(element.object(), osmdata, progress);
+            ctx.apply(element.object());
         }
 
         auto *source = element.data_source();
         if (source->next()) {
             queue.emplace(source->get(), source);
         }
     }
+    ctx.eof();
 
     for (auto &data_source : data_sources) {
         data_source.close();
@@ -295,7 +324,4 @@ void process_files(std::vector<osmium::io::File> const &files,
     } else {
         process_multiple_files(files, osmdata, progress, append);
     }
-
-    osmdata.after_relations();
-    progress.print_summary();
 }

src/middle-pgsql.cpp

@@ -26,6 +26,7 @@
 #include "node-ram-cache.hpp"
 #include "options.hpp"
 #include "osmtypes.hpp"
+#include "pgsql-helper.hpp"
 #include "util.hpp"
 
 /**
@@ -314,6 +315,40 @@ std::size_t middle_query_pgsql_t::get_way_node_locations_db(
     return count;
 }
 
+void middle_pgsql_t::node(osmium::Node const &node)
+{
+    if (node.deleted()) {
+        node_delete(node.id());
+    } else {
+        if (m_options->append) {
+            node_delete(node.id());
+        }
+        node_set(node);
+    }
+}
+
+void middle_pgsql_t::way(osmium::Way const &way) {
+    if (way.deleted()) {
+        way_delete(way.id());
+    } else {
+        if (m_options->append) {
+            way_delete(way.id());
+        }
+        way_set(way);
+    }
+}
+
+void middle_pgsql_t::relation(osmium::Relation const &relation) {
+    if (relation.deleted()) {
+        relation_delete(relation.id());
+    } else {
+        if (m_options->append) {
+            relation_delete(relation.id());
+        }
+        relation_set(relation);
+    }
+}
+
 void middle_pgsql_t::node_set(osmium::Node const &node)
 {
     m_cache->set(node.id(), node.location());
@@ -567,13 +602,32 @@ void middle_pgsql_t::relation_delete(osmid_t osm_id)
     m_db_copy.delete_object(osm_id);
 }
 
-void middle_pgsql_t::analyze()
+void middle_pgsql_t::after_nodes()
 {
-    for (auto const &table : m_tables) {
-        m_db_connection.exec("ANALYZE {}"_format(table.name()));
+    m_db_copy.sync();
+    if (m_options->flat_node_file.empty()) {
+        auto const &table = m_tables[NODE_TABLE];
+        analyze_table(m_db_connection, table.schema(), table.name());
     }
 }
 
+void middle_pgsql_t::after_ways()
+{
+    m_db_copy.sync();
+    auto const &table = m_tables[WAY_TABLE];
+    analyze_table(m_db_connection, table.schema(), table.name());
+}
+
+void middle_pgsql_t::after_relations()
+{
+    m_db_copy.sync();
+    auto const &table = m_tables[REL_TABLE];
+    analyze_table(m_db_connection, table.schema(), table.name());
+
+    // release the copy thread and its database connection
+    m_copy_thread->finish();
+}
+
 middle_query_pgsql_t::middle_query_pgsql_t(
     std::string const &conninfo, std::shared_ptr<node_ram_cache> const &cache,
     std::shared_ptr<node_persistent_cache> const &persistent_cache)
@@ -611,15 +665,6 @@ void middle_pgsql_t::start()
     }
 }
 
-void middle_pgsql_t::commit()
-{
-    m_db_copy.sync();
-    // release the copy thread and its query connection
-    m_copy_thread->finish();
-}
-
-void middle_pgsql_t::flush() { m_db_copy.sync(); }
-
 void middle_pgsql_t::stop(thread_pool_t &pool)
 {
     m_cache.reset();

src/middle-pgsql.hpp

@@ -54,25 +54,20 @@ struct table_sql {
     char const *create_fw_dep_indexes = "";
 };
 
-struct middle_pgsql_t : public slim_middle_t
+struct middle_pgsql_t : public middle_t
 {
     explicit middle_pgsql_t(options_t const *options);
 
     void start() override;
     void stop(thread_pool_t &pool) override;
-    void analyze() override;
-    void commit() override;
 
-    void node_set(osmium::Node const &node) override;
-    void node_delete(osmid_t id) override;
+    void node(osmium::Node const &node) override;
+    void way(osmium::Way const &way) override;
+    void relation(osmium::Relation const &rel) override;
 
-    void way_set(osmium::Way const &way) override;
-    void way_delete(osmid_t id) override;
-
-    void relation_set(osmium::Relation const &rel) override;
-    void relation_delete(osmid_t id) override;
-
-    void flush() override;
+    void after_nodes() override;
+    void after_ways() override;
+    void after_relations() override;
 
     idlist_t get_ways_by_node(osmid_t osm_id) override;
     idlist_t get_rels_by_node(osmid_t osm_id) override;
@@ -122,6 +117,15 @@ struct middle_pgsql_t : public slim_middle_t
         NUM_TABLES
     };
 
+    void node_set(osmium::Node const &node);
+    void node_delete(osmid_t id);
+
+    void way_set(osmium::Way const &way);
+    void way_delete(osmid_t id);
+
+    void relation_set(osmium::Relation const &rel);
+    void relation_delete(osmid_t id);
+
     void buffer_store_tags(osmium::OSMObject const &obj, bool attrs);
 
     idlist_t get_ids(const char* stmt, osmid_t osm_id);

src/middle-ram.cpp

@@ -30,18 +30,21 @@
  *
  */
 
-void middle_ram_t::node_set(osmium::Node const &node)
+void middle_ram_t::node(osmium::Node const &node)
 {
+    assert(node.visible());
     m_cache->set(node.id(), node.location());
 }
 
-void middle_ram_t::way_set(osmium::Way const &way)
+void middle_ram_t::way(osmium::Way const &way)
 {
+    assert(way.visible());
     m_ways.set(way.id(), new ramWay{way, m_extra_attributes});
 }
 
-void middle_ram_t::relation_set(osmium::Relation const &rel)
+void middle_ram_t::relation(osmium::Relation const &rel)
 {
+    assert(rel.visible());
     m_rels.set(rel.id(), new ramRel{rel, m_extra_attributes});
 }
 

src/middle-ram.hpp

@@ -96,22 +96,19 @@ struct middle_ram_t : public middle_t, public middle_query_t
 
     void start() override {}
     void stop(thread_pool_t &pool) override;
-    void analyze() override {}
-    void commit() override {}
 
-    void node_set(osmium::Node const &node) override;
+    void node(osmium::Node const &node) override;
+    void way(osmium::Way const &way) override;
+    void relation(osmium::Relation const &rel) override;
+
     size_t nodes_get_list(osmium::WayNodeList *nodes) const override;
 
-    void way_set(osmium::Way const &way) override;
     bool way_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;
 
     bool relation_get(osmid_t id,
                       osmium::memory::Buffer &buffer) const override;
-    void relation_set(osmium::Relation const &rel) override;
-
-    void flush() override {}
 
     std::shared_ptr<middle_query_t> get_query_instance() override;
 

src/middle.cpp

@@ -0,0 +1,15 @@
+
+#include "middle-pgsql.hpp"
+#include "middle-ram.hpp"
+#include "middle.hpp"
+#include "options.hpp"
+
+std::shared_ptr<middle_t> create_middle(options_t const &options)
+{
+    if (options.slim) {
+        return std::make_shared<middle_pgsql_t>(&options);
+    }
+
+    return std::make_shared<middle_ram_t>(&options);
+}
+