Work around PostgreSQL transaction poisoning in schema version query

Note that this only works in 9.4+. For older versions the workaround is
to "pre-call" database::schema_version() outside of any transaction.

3 files changed, 115 insertions, 24 deletions

diff --git a/odb/pgsql/connection.cxx b/odb/pgsql/connection.cxx
index 1939e23..01399c9 100644
--- a/odb/pgsql/connection.cxx
+++ b/odb/pgsql/connection.cxx
@@ -77,6 +77,12 @@ namespace odb
       clear_prepared_map ();
     }
 
+    int connection::
+    server_version () const
+    {
+      return PQserverVersion (handle_);
+    }
+
     transaction_impl* connection::
     begin ()
     {
diff --git a/odb/pgsql/connection.hxx b/odb/pgsql/connection.hxx
index 57837ee..316dfeb 100644
--- a/odb/pgsql/connection.hxx
+++ b/odb/pgsql/connection.hxx
@@ -115,6 +115,12 @@ namespace odb
         return handle_;
       }
 
+      // Server version as returned by PQserverVersion(), for example, 90200
+      // (9.2.0), 90201 (9.2.1), 100000 (10.0), 110001 (11.1).
+      //
+      int
+      server_version () const;
+
       statement_cache_type&
       statement_cache ()
       {
diff --git a/odb/pgsql/database.cxx b/odb/pgsql/database.cxx
index 630150c..e7f3bd0 100644
--- a/odb/pgsql/database.cxx
+++ b/odb/pgsql/database.cxx
@@ -2,6 +2,7 @@
 // copyright : Copyright (c) 2009-2017 Code Synthesis Tools CC
 // license   : GNU GPL v2; see accompanying LICENSE file
 
+#include <cstring> // strlen()
 #include <sstream>
 
 #include <odb/pgsql/traits.hxx>
@@ -234,17 +235,17 @@ namespace odb
     {
       schema_version_info& svi (schema_version_map_[name]);
 
+      // Quoted table name.
+      //
+      const char* table (
+        !svi.version_table.empty ()     ? svi.version_table.c_str ()     :
+        !schema_version_table_.empty () ? schema_version_table_.c_str () :
+        /*                             */ "\"schema_version\"");
+
       // Construct the SELECT statement text.
       //
       string text ("SELECT \"version\", \"migration\" FROM ");
-
-      if (!svi.version_table.empty ())
-        text += svi.version_table; // Already quoted.
-      else if (!schema_version_table_.empty ())
-        text += schema_version_table_; // Already quoted.
-      else
-        text += "\"schema_version\"";
-
+      text += table;
       text += " WHERE \"name\" = $1";
 
       // Bind parameters and results.
@@ -284,11 +285,50 @@ namespace odb
       pgsql::connection& c (
         cp != 0 ? *cp : transaction::current ().connection ());
 
-      try
+      // If we are in the user's transaction then things are complicated. When
+      // we try to execute SELECT on a non-existent table, PG "poisons" the
+      // transaction (those "current transaction is aborted, commands ignored
+      // until end of transaction block" messages in the log). Which means all
+      // the user's schema creation statements that are likely to follow will
+      // fail.
+      //
+      // There doesn't seem to be a better way to solve this than to check for
+      // the table's existence. It is relatively easy to do with to_regclass()
+      // in 9.4+ and a real pain in earlier versions. So we are going to do
+      // this for 9.4+ and for older versions the workaround is to "pre-call"
+      // database::schema_version() outside of any transaction.
+      //
+      bool exists (true);
+      if (cp == 0 && c.server_version () >= 90400)
       {
+        size_t psize[1] = {strlen (table)};
+        bool pnull[1] = {false};
+        bind pbind[1] = {{bind::text,
+                          const_cast<char*> (table),
+                          &psize[0],
+                          psize[0],
+                          &pnull[0],
+                          0}};
+        binding param (pbind, 1);
+        param.version++;
+
+        unsigned int param_types[1] = {text_oid};
+
+        char* values[1];
+        int lengths[1];
+        int formats[1];
+        native_binding nparam (values, lengths, formats, 1);
+
+        bool rnull[1];
+        bind rbind[1] = {{bind::boolean_, &exists, 0, 0, &rnull[1], 0}};
+        binding result (rbind, 1);
+        result.version++;
+
+        // Note that to_regclass() seems happy to accept a quoted table name.
+        //
         select_statement st (c,
-                             "odb_database_schema_version",
-                             text.c_str (),
+                             "odb_database_schema_version_exists",
+                             "SELECT to_regclass($1::cstring) IS NOT NULL",
                              false, // Don't process.
                              false, // Don't optimize.
                              param_types,
@@ -297,6 +337,7 @@ namespace odb
                              nparam,
                              result,
                              false);
+
         st.execute ();
         auto_result ar (st);
 
@@ -304,16 +345,10 @@ namespace odb
         {
         case select_statement::success:
           {
-            value_traits<unsigned long long, id_bigint>::set_value (
-              svi.version, version, rnull[0]);
             assert (st.fetch () == select_statement::no_data);
             break;
           }
         case select_statement::no_data:
-          {
-            svi.version = 0; // No schema.
-            break;
-          }
         case select_statement::truncated:
           {
             assert (false);
@@ -321,14 +356,58 @@ namespace odb
           }
         }
       }
-      catch (const database_exception& e)
+
+      // Assume no schema until determined otherwise.
+      //
+      svi.version = 0;
+
+      if (exists)
       {
-        // Detect the case where there is no version table.
-        //
-        if (e.sqlstate () == "42P01")
-          svi.version = 0; // No schema.
-        else
-          throw;
+        try
+        {
+          select_statement st (c,
+                               "odb_database_schema_version_query",
+                               text.c_str (),
+                               false, // Don't process.
+                               false, // Don't optimize.
+                               param_types,
+                               1,
+                               param,
+                               nparam,
+                               result,
+                               false);
+          st.execute ();
+          auto_result ar (st);
+
+          switch (st.fetch ())
+          {
+          case select_statement::success:
+            {
+              value_traits<unsigned long long, id_bigint>::set_value (
+                svi.version, version, rnull[0]);
+              assert (st.fetch () == select_statement::no_data);
+              break;
+            }
+          case select_statement::no_data:
+            {
+              // No schema.
+              break;
+            }
+          case select_statement::truncated:
+            {
+              assert (false);
+              break;
+            }
+          }
+        }
+        catch (const database_exception& e)
+        {
+          // Detect the case where there is no version table (the implicit
+          // transaction case).
+          //
+          if (e.sqlstate () != "42P01")
+            throw;
+        }
       }
 
       return svi;