Add grouped-linemerge generalization

CMakeLists.txt

@@ -264,6 +264,7 @@ else()
                 src/gen/gen-base.cpp
                 src/gen/gen-create.cpp
                 src/gen/gen-discrete-isolation.cpp
+                src/gen/gen-grouped-linemerge.cpp
                 src/gen/gen-rivers.cpp
                 src/gen/gen-tile-builtup.cpp
                 src/gen/gen-tile-raster.cpp

flex-config/gen/grouped-linemerge.lua

@@ -0,0 +1,106 @@
+-- This config example file is released into the Public Domain.
+--
+-- This Lua config demonstrates the 'grouped-linemerge' generalization. It
+-- merges connected lines that share the same set of grouping columns into
+-- single (multi-)lines, the equivalent of
+--
+--   SELECT cols..., (ST_Dump(ST_LineMerge(ST_Collect(geom)))).geom
+--     FROM roads GROUP BY cols...
+--
+-- but done globally and maintained incrementally on updates. A typical use is
+-- merging road segments that render identically (same name/ref/highway/layer)
+-- so that labels and route shields are placed on the whole road instead of on
+-- each individual OSM way, without the artifacts you get when merging only
+-- within a tile.
+--
+-- NOTE THAT THE GENERALIZATION SUPPORT IS EXPERIMENTAL AND MIGHT CHANGE
+-- WITHOUT NOTICE!
+--
+-- Workflow:
+--   * Import as usual:           osm2pgsql -O flex -S grouped-linemerge.lua DATA.osm.pbf
+--   * Build the merged table:    osm2pgsql-gen -S grouped-linemerge.lua
+--   * Apply an update:           osm2pgsql -a -O flex -S grouped-linemerge.lua CHANGES.osc.gz
+--   * Update the merged table:   osm2pgsql-gen -a -S grouped-linemerge.lua
+
+-- An expire output with an 'endpoint_table' records the exact endpoints (start
+-- and end point) of every way added/edited/deleted during an update, as POINT
+-- rows. The grouped-linemerge generalization consumes these points: it walks
+-- each affected connected component out from the changed endpoints and
+-- re-merges only those, matching by exact endpoint equality (no tiles, no
+-- area scan). Deletes contribute the old way's endpoints automatically.
+local exp_roads = osm2pgsql.define_expire_output({
+    endpoint_table = 'exp_roads_endpoints',
+})
+
+-- The source table with the original road segments (one row per OSM way).
+local roads = osm2pgsql.define_table({
+    name = 'roads',
+    ids = { type = 'way', id_column = 'way_id' },
+    columns = {
+        { column = 'name', type = 'text' },
+        { column = 'ref', type = 'text' },
+        { column = 'highway', type = 'text' },
+        { column = 'layer', type = 'int' },
+        -- Attach the expire output to the geometry so that any change to a
+        -- road's geometry (add/modify/delete) records its endpoints.
+        { column = 'geom', type = 'linestring', not_null = true,
+            expire = { { output = exp_roads } } },
+    }
+})
+
+-- The destination table with the merged roads. Its columns are exactly the
+-- grouping columns plus the geometry. It has no OSM id column (it is derived
+-- data maintained by osm2pgsql-gen, not by the normal update process); the
+-- warning osm2pgsql prints about that is expected.
+osm2pgsql.define_table({
+    name = 'roads_merged',
+    columns = {
+        { column = 'name', type = 'text' },
+        { column = 'ref', type = 'text' },
+        { column = 'highway', type = 'text' },
+        { column = 'layer', type = 'int' },
+        { column = 'geom', type = 'linestring', not_null = true },
+    }
+})
+
+function osm2pgsql.process_way(object)
+    local highway = object.tags.highway
+    if not highway then
+        return
+    end
+    roads:insert({
+        name = object.tags.name,
+        ref = object.tags.ref,
+        highway = highway,
+        layer = tonumber(object.tags.layer),
+        geom = object:as_linestring(),
+    })
+end
+
+function osm2pgsql.process_gen()
+    osm2pgsql.run_gen('grouped-linemerge', {
+        name = 'roads',          -- name (for logging)
+        debug = false,           -- set to true for more detailed debug output
+        src_table = 'roads',     -- input table with the line segments
+        dest_table = 'roads_merged', -- output table for the merged lines
+        geom_column = 'geom',    -- geometry column (same in src and dest)
+
+        -- Lines are merged when ALL of these columns are equal (NULLs compare
+        -- equal). Pass them as a comma-separated list.
+        group_by_columns = 'name, ref, highway, layer',
+
+        -- Optional pre-filter (SQL boolean expression on the source columns).
+        -- Lines not matching are completely excluded from the generalization.
+        -- Here we only merge roads that carry a label or a shield.
+        where = 'name IS NOT NULL OR ref IS NOT NULL',
+
+        -- In append mode, the table of exact changed-way endpoints to consume
+        -- (written by the expire output's 'endpoint_table' above).
+        endpoint_table = 'exp_roads_endpoints',
+
+        -- Create functional endpoint indexes on the src/dest tables in create
+        -- mode. These make the incremental component walk fast. Set to false
+        -- if you manage the indexes yourself.
+        create_indexes = true,
+    })
+end

src/expire-output.cpp

@@ -10,9 +10,12 @@
 #include "expire-output.hpp"
 
 #include "format.hpp"
+#include "geom.hpp"
+#include "hex.hpp"
 #include "logging.hpp"
 #include "pgsql.hpp"
 #include "tile.hpp"
+#include "wkb.hpp"
 
 #include <cerrno>
 #include <system_error>
@@ -50,10 +53,43 @@ void expire_output_t::add_tiles(
     m_tiles.insert(dirty_tiles.cbegin(), dirty_tiles.cend());
 }
 
+void expire_output_t::add_endpoints(geom::geometry_t const &geom)
+{
+    auto const add_line = [this](geom::linestring_t const &line) {
+        if (line.empty()) {
+            return;
+        }
+        auto const &first = line.front();
+        auto const &last = line.back();
+        m_endpoints.emplace(first.x(), first.y());
+        m_endpoints.emplace(last.x(), last.y());
+    };
+
+    std::lock_guard<std::mutex> const guard{*m_tiles_mutex};
+
+    m_endpoint_srid = geom.srid();
+    if (geom.is_linestring()) {
+        add_line(geom.get<geom::linestring_t>());
+    } else if (geom.is_multilinestring()) {
+        for (auto const &line : geom.get<geom::multilinestring_t>()) {
+            add_line(line);
+        }
+    }
+}
+
 bool expire_output_t::empty() noexcept
 {
     std::lock_guard<std::mutex> const guard{*m_tiles_mutex};
-    return m_tiles.empty();
+    return m_tiles.empty() && m_endpoints.empty();
+}
+
+std::vector<std::pair<double, double>> expire_output_t::get_endpoints()
+{
+    std::lock_guard<std::mutex> const guard{*m_tiles_mutex};
+    std::vector<std::pair<double, double>> endpoints(m_endpoints.cbegin(),
+                                                     m_endpoints.cend());
+    m_endpoints.clear();
+    return endpoints;
 }
 
 quadkey_list_t expire_output_t::get_tiles()
@@ -79,6 +115,9 @@ expire_output_t::output(connection_params_t const &connection_params)
     if (!m_table.empty()) {
         num = output_tiles_to_table(get_tiles(), connection_params);
     }
+    if (!m_endpoint_table.empty()) {
+        output_endpoints_to_table(get_endpoints(), connection_params);
+    }
     return num;
 }
 
@@ -140,16 +179,49 @@ std::size_t expire_output_t::output_tiles_to_table(
     return count;
 }
 
+std::size_t expire_output_t::output_endpoints_to_table(
+    std::vector<std::pair<double, double>> const &endpoints,
+    connection_params_t const &connection_params) const
+{
+    if (endpoints.empty()) {
+        return 0;
+    }
+
+    auto const qn = qualified_name(m_schema, m_endpoint_table);
+
+    pg_conn_t const db_connection{connection_params, "expire"};
+
+    db_connection.prepare("insert_endpoints",
+                          "INSERT INTO {} (geom) VALUES ($1::geometry)", qn);
+
+    for (auto const &[x, y] : endpoints) {
+        geom::geometry_t const point{geom::point_t{x, y}, m_endpoint_srid};
+        db_connection.exec_prepared("insert_endpoints",
+                                    util::encode_hex(geom_to_ewkb(point)));
+    }
+
+    return endpoints.size();
+}
+
 void expire_output_t::create_output_table(pg_conn_t const &db_connection) const
 {
-    auto const qn = qualified_name(m_schema, m_table);
-    db_connection.exec(
-        "CREATE TABLE IF NOT EXISTS {} ("
-        " zoom int4 NOT NULL,"
-        " x int4 NOT NULL,"
-        " y int4 NOT NULL,"
-        " first timestamp with time zone DEFAULT CURRENT_TIMESTAMP(0),"
-        " last timestamp with time zone DEFAULT CURRENT_TIMESTAMP(0),"
-        " PRIMARY KEY (zoom, x, y))",
-        qn);
+    if (!m_table.empty()) {
+        auto const qn = qualified_name(m_schema, m_table);
+        db_connection.exec(
+            "CREATE TABLE IF NOT EXISTS {} ("
+            " zoom int4 NOT NULL,"
+            " x int4 NOT NULL,"
+            " y int4 NOT NULL,"
+            " first timestamp with time zone DEFAULT CURRENT_TIMESTAMP(0),"
+            " last timestamp with time zone DEFAULT CURRENT_TIMESTAMP(0),"
+            " PRIMARY KEY (zoom, x, y))",
+            qn);
+    }
+
+    if (!m_endpoint_table.empty()) {
+        auto const qn = qualified_name(m_schema, m_endpoint_table);
+        db_connection.exec("CREATE TABLE IF NOT EXISTS {} ("
+                           " geom geometry(Point) NOT NULL)",
+                           qn);
+    }
 }

src/expire-output.hpp

@@ -17,16 +17,22 @@
 #include <cstdint>
 #include <memory>
 #include <mutex>
+#include <set>
 #include <string>
 #include <unordered_set>
 #include <utility>
+#include <vector>
 
 constexpr std::size_t DEFAULT_MAX_TILES_GEOMETRY = 10'000'000;
 constexpr std::size_t DEFAULT_MAX_TILES_OVERALL = 50'000'000;
 
 class pg_conn_t;
 class connection_params_t;
 
+namespace geom {
+class geometry_t;
+} // namespace geom
+
 /**
  * Output for tile expiry.
  */
@@ -53,6 +59,35 @@ class expire_output_t
         m_table = std::move(table);
     }
 
+    std::string const &endpoint_table() const noexcept
+    {
+        return m_endpoint_table;
+    }
+
+    void set_endpoint_table(std::string table)
+    {
+        m_endpoint_table = std::move(table);
+    }
+
+    /// Does this output write expired tiles (to a file and/or table)?
+    bool has_tile_output() const noexcept
+    {
+        return !m_filename.empty() || !m_table.empty();
+    }
+
+    /// Does this output write the endpoints of changed geometries to a table?
+    bool has_endpoint_output() const noexcept
+    {
+        return !m_endpoint_table.empty();
+    }
+
+    /**
+     * Record the endpoints (first and last point of each linestring) of a
+     * changed geometry. They are written to the endpoint table on output.
+     * Non-(multi)linestring geometries contribute no endpoints. Thread-safe.
+     */
+    void add_endpoints(geom::geometry_t const &geom);
+
     uint32_t minzoom() const noexcept { return m_minzoom; }
     void set_minzoom(uint32_t minzoom) noexcept { m_minzoom = minzoom; }
 
@@ -116,6 +151,19 @@ class expire_output_t
     output_tiles_to_table(quadkey_list_t const &tiles_at_maxzoom,
                           connection_params_t const &connection_params) const;
 
+    /// Take and clear the collected endpoints. Thread-safe.
+    std::vector<std::pair<double, double>> get_endpoints();
+
+    /**
+     * Write the collected endpoints as POINT geometries to the endpoint table.
+     *
+     * \param endpoints The endpoint coordinates to write
+     * \param connection_params Database connection parameters
+     */
+    std::size_t
+    output_endpoints_to_table(std::vector<std::pair<double, double>> const &endpoints,
+                              connection_params_t const &connection_params) const;
+
     /**
      * Access to the m_tiles collection of expired tiles must go through
      * this mutex, because it can happend from several threads at the same
@@ -136,6 +184,16 @@ class expire_output_t
     /// The table (if any) for output
     std::string m_table;
 
+    /// The table (if any) for changed-geometry endpoints
+    std::string m_endpoint_table;
+
+    /// Collected endpoints (deduplicated) of changed geometries. Guarded by
+    /// m_tiles_mutex (same access pattern as m_tiles).
+    std::set<std::pair<double, double>> m_endpoints;
+
+    /// SRID of the collected endpoints (all geometries share the table's SRID).
+    int m_endpoint_srid = 0;
+
     /// Minimum zoom level for output
     uint32_t m_minzoom = 0;
 

src/flex-lua-expire-output.cpp

@@ -39,10 +39,20 @@ create_expire_output(lua_State *lua_state, std::string const &default_schema,
     new_expire_output.set_schema_and_table(schema, table);
     lua_pop(lua_state, 2); // "schema" and "table"
 
+    // optional "endpoint_table" field: write the endpoints of changed
+    // geometries as POINT rows (in the same schema) instead of, or in addition
+    // to, expired tiles.
+    auto const *endpoint_table = luaX_get_table_string(
+        lua_state, "endpoint_table", -1, "The expire output", "");
+    check_identifier(endpoint_table, "endpoint_table field");
+    new_expire_output.set_endpoint_table(endpoint_table);
+    lua_pop(lua_state, 1); // "endpoint_table"
+
     if (new_expire_output.filename().empty() &&
-        new_expire_output.table().empty()) {
-        throw std::runtime_error{
-            "Must set 'filename' and/or 'table' on expire output."};
+        new_expire_output.table().empty() &&
+        new_expire_output.endpoint_table().empty()) {
+        throw std::runtime_error{"Must set 'filename', 'table', and/or "
+                                 "'endpoint_table' on expire output."};
     }
 
     // required "maxzoom" field