Finish schema support in flex output

This completes the schema support for the flex output.

You have to define a schema before using it. osm2pgsql might not have
the needed permissions for creating a schema, so it will not try to
create it.

Author: joto

docs/flex.md

@@ -69,6 +69,8 @@ With the `osm2pgsql.define_table()` function you can also define tables that
   additional column.
 * are in a specific PostgresSQL tablespace (set `data_tablespace`) or that
   get their indexes created in a specific tablespace (set `index_tablespace`).
+* are in a specific schema (set `schema`). Note that the schema has to be
+  created before you start osm2pgsql.
 
 If you are using the `osm2pgsql.define_(node|way|relation|area)_table()`
 convenience functions, osm2pgsql will automatically create an id column named

flex-config/with-schema.lua

@@ -0,0 +1,22 @@
+
+-- This configuration for the flex output shows how to define a table in
+-- a PostgreSQL schema.
+
+local dtable = osm2pgsql.define_table{
+    name = "data",
+    schema = "myschema",
+    ids = { type = 'way', id_column = 'way_id' },
+    columns = {
+        { column = 'attrs', type = 'hstore' },
+        { column = 'tags',  type = 'hstore' },
+        { column = 'geom',  type = 'geometry' },
+    }
+}
+
+function osm2pgsql.process_way(object)
+    dtable:add_row({
+        tags = object.tags,
+        geom = { create = 'line' }
+    })
+end
+

src/db-copy.cpp

@@ -177,12 +177,13 @@ void db_copy_thread_t::thread_t::start_copy(
 {
     assert(!m_inflight);
 
+    auto const qname = qualified_name(target->schema, target->name);
     fmt::memory_buffer sql;
-    sql.reserve(target->name.size() + target->rows.size() + 14);
+    sql.reserve(qname.size() + target->rows.size() + 20);
     if (target->rows.empty()) {
-        fmt::format_to(sql, FMT_STRING("COPY {} FROM STDIN"), target->name);
+        fmt::format_to(sql, FMT_STRING("COPY {} FROM STDIN"), qname);
     } else {
-        fmt::format_to(sql, FMT_STRING("COPY {} ({}) FROM STDIN"), target->name,
+        fmt::format_to(sql, FMT_STRING("COPY {} ({}) FROM STDIN"), qname,
                        target->rows);
     }
 

src/db-copy.hpp

@@ -11,14 +11,15 @@
 #include <vector>
 
 #include "osmtypes.hpp"
-
-class pg_conn_t;
+#include "pgsql.hpp"
 
 /**
  * Table information necessary for building SQL queries.
  */
 struct db_target_descr_t
 {
+    /// Schema of the target table (can be empty for default schema)
+    std::string schema;
     /// Name of the target table for the copy operation.
     std::string name;
     /// Comma-separated list of rows for copy operation (when empty: all rows)
@@ -31,7 +32,8 @@ struct db_target_descr_t
      */
     bool same_copy_target(db_target_descr_t const &other) const noexcept
     {
-        return (this == &other) || (name == other.name && rows == other.rows);
+        return (this == &other) || (schema == other.schema &&
+                                    name == other.name && rows == other.rows);
     }
 
     db_target_descr_t() = default;
@@ -190,7 +192,8 @@ class db_cmd_copy_delete_t : public db_cmd_copy_t
     void delete_data(pg_conn_t *conn) override
     {
         if (m_deleter.has_data()) {
-            m_deleter.delete_rows(target->name, target->id, conn);
+            m_deleter.delete_rows(qualified_name(target->schema, target->name),
+                                  target->id, conn);
         }
     }
 

src/flex-table.cpp

@@ -45,12 +45,12 @@ std::string flex_table_t::id_column_names() const
 
 std::string flex_table_t::full_name() const
 {
-    return "\"" + schema() + "\".\"" + name() + "\"";
+    return qualified_name(schema(), name());
 }
 
 std::string flex_table_t::full_tmp_name() const
 {
-    return "\"" + schema() + "\".\"" + name() + "_tmp\"";
+    return qualified_name(schema(), name() + "_tmp");
 }
 
 flex_table_column_t &flex_table_t::add_column(std::string const &name,
@@ -162,11 +162,6 @@ void table_connection_t::start(bool append)
     m_db_connection->exec("RESET client_min_messages");
 
     if (!append) {
-        if (table().schema() != "public") {
-            m_db_connection->exec(
-                "CREATE SCHEMA IF NOT EXISTS \"{}\""_format(table().schema()));
-        }
-
         m_db_connection->exec(table().build_sql_create_table(
             table().has_geom_column() ? flex_table_t::table_type::interim
                                       : flex_table_t::table_type::permanent,
@@ -299,7 +294,7 @@ void table_connection_t::stop(bool updateable, bool append)
     }
 
     fmt::print(stderr, "Analyzing table '{}'...\n", table().name());
-    m_db_connection->exec("ANALYZE \"{}\""_format(table().name()));
+    m_db_connection->exec("ANALYZE " + table().full_name());
 
     fmt::print(stderr, "All postprocessing on table '{}' done in {}s.\n",
                table().name(), timer.stop());

src/flex-table.hpp

@@ -199,7 +199,9 @@ class table_connection_t
                           table->name(), table->id_column_names(),
                           table->build_sql_column_list())),
       m_copy_mgr(copy_thread), m_db_connection(nullptr)
-    {}
+    {
+        m_target->schema = table->schema();
+    }
 
     void connect(std::string const &conninfo);
 

src/output-flex.cpp

@@ -593,18 +593,18 @@ flex_table_t &output_flex_t::create_flex_table()
     // optional "data_tablespace" field
     lua_getfield(lua_state(), -1, "data_tablespace");
     if (lua_isstring(lua_state(), -1)) {
-        std::string const schema = lua_tostring(lua_state(), -1);
-        check_name(schema, "data_tablespace");
-        new_table.set_data_tablespace(schema);
+        std::string const tablespace = lua_tostring(lua_state(), -1);
+        check_name(tablespace, "data_tablespace");
+        new_table.set_data_tablespace(tablespace);
     }
     lua_pop(lua_state(), 1);
 
     // optional "index_tablespace" field
     lua_getfield(lua_state(), -1, "index_tablespace");
     if (lua_isstring(lua_state(), -1)) {
-        std::string const schema = lua_tostring(lua_state(), -1);
-        check_name(schema, "index_tablespace");
-        new_table.set_index_tablespace(schema);
+        std::string const tablespace = lua_tostring(lua_state(), -1);
+        check_name(tablespace, "index_tablespace");
+        new_table.set_index_tablespace(tablespace);
     }
     lua_pop(lua_state(), 1);
 

src/pgsql.cpp

@@ -149,10 +149,28 @@ std::string tablespace_clause(std::string const &name)
     std::string sql;
 
     if (!name.empty()) {
-        sql += " TABLESPACE ";
+        sql += " TABLESPACE \"";
         sql += name;
+        sql += '"';
     }
 
     return sql;
 }
 
+std::string qualified_name(std::string const &schema, std::string const &name)
+{
+    std::string result{"\""};
+
+    if (!schema.empty()) {
+        result.reserve(schema.size() + name.size() + 5);
+        result += schema;
+        result += "\".\"";
+    } else {
+        result.reserve(name.size() + 2);
+    }
+
+    result += name;
+    result += '"';
+
+    return result;
+}

src/pgsql.hpp

@@ -142,6 +142,16 @@ class pg_conn_t
     std::unique_ptr<PGconn, pg_conn_deleter_t> m_conn;
 };
 
+/**
+ * Return a TABLESPACE clause with the specified tablespace name or an empty
+ * string if the name is empty.
+ */
 std::string tablespace_clause(std::string const &name);
 
+/**
+ * Return the possibly schema-qualified name of a table. Names are enclosed
+ * in double quotes.
+ */
+std::string qualified_name(std::string const &schema, std::string const &name);
+
 #endif // OSM2PGSQL_PGSQL_HPP

tests/CMakeLists.txt

@@ -60,6 +60,7 @@ set_test(test-output-pgsql-validgeom)
 set_test(test-output-pgsql-z_order)
 set_test(test-parse-osmium)
 set_test(test-persistent-cache LABELS NoDB)
+set_test(test-pgsql)
 set_test(test-ram-cache LABELS NoDB)
 set_test(test-taginfo LABELS NoDB)
 set_test(test-util LABELS NoDB)