Executor: Add migration functionality. Tests: add IntTest + FloatTest.

Author: lganzzzo

src/CMakeLists.txt

@@ -24,7 +24,7 @@ add_library(${OATPP_THIS_MODULE_NAME}
         oatpp-postgresql/QueryResult.cpp
         oatpp-postgresql/QueryResult.hpp
         oatpp-postgresql/Types.hpp
-)
+        oatpp-postgresql/orm.hpp)
 
 set_target_properties(${OATPP_THIS_MODULE_NAME} PROPERTIES
         CXX_STANDARD 11

src/oatpp-postgresql/Executor.cpp

@@ -29,12 +29,31 @@
 
 #include "QueryResult.hpp"
 
+#include "oatpp/orm/Transaction.hpp"
+
 #include "oatpp/core/data/stream/ChunkedBuffer.hpp"
+#include "oatpp/core/macro/codegen.hpp"
 
 #include <vector>
 
 namespace oatpp { namespace postgresql {
 
+namespace {
+
+#include OATPP_CODEGEN_BEGIN(DTO)
+
+class VersionRow : public oatpp::DTO {
+
+  DTO_INIT(VersionRow, DTO);
+
+  DTO_FIELD(Int64, version);
+
+};
+
+#include OATPP_CODEGEN_END(DTO)
+
+}
+
 Executor::QueryParams::QueryParams(const StringTemplate& queryTemplate,
                                    const std::unordered_map<oatpp::String, oatpp::Void>& params,
                                    const mapping::TypeMapper& typeMapper,
@@ -44,7 +63,11 @@ Executor::QueryParams::QueryParams(const StringTemplate& queryTemplate,
   auto extra = std::static_pointer_cast<ql_template::Parser::TemplateExtra>(queryTemplate.getExtraData());
 
   query = extra->preparedTemplate->c_str();
-  queryName = extra->templateName->c_str();
+  if(extra->templateName) {
+    queryName = extra->templateName->c_str();
+  } else {
+    queryName = nullptr;
+  }
 
   count = queryTemplate.getTemplateVariables().size();
 
@@ -214,43 +237,188 @@ std::shared_ptr<orm::QueryResult> Executor::execute(const StringTemplate& queryT
 
 }
 
-std::shared_ptr<orm::QueryResult> Executor::begin(const std::shared_ptr<orm::Connection>& connection) {
+std::shared_ptr<orm::QueryResult> Executor::exec(const oatpp::String& statement,
+                                                 const std::shared_ptr<orm::Connection>& connection,
+                                                 bool useExecParams)
+{
 
   std::shared_ptr<orm::Connection> conn = connection;
   if(!conn) {
     conn = getConnection();
   }
 
   auto pgConnection = std::static_pointer_cast<postgresql::Connection>(conn);
-  PGresult *qres = PQexec(pgConnection->getHandle(), "BEGIN");
+
+  PGresult *qres;
+  if(useExecParams) {
+    qres = PQexecParams(pgConnection->getHandle(),
+                        statement->c_str(),
+                        0 /* nParams */,
+                        nullptr /* paramTypes */,
+                        nullptr /* paramValues */,
+                        nullptr /* paramLengths */,
+                        nullptr /* paramFormats */,
+                        1 /* resultFormat */);
+  } else {
+    qres = PQexec(pgConnection->getHandle(), statement->c_str());
+  }
+
   return std::make_shared<QueryResult>(qres, pgConnection, m_connectionProvider, m_resultMapper);
 
 }
 
+std::shared_ptr<orm::QueryResult> Executor::begin(const std::shared_ptr<orm::Connection>& connection) {
+  return exec("BEGIN", connection);
+}
+
 std::shared_ptr<orm::QueryResult> Executor::commit(const std::shared_ptr<orm::Connection>& connection) {
-  auto pgConnection = std::static_pointer_cast<postgresql::Connection>(connection);
-  PGresult *qres = PQexec(pgConnection->getHandle(), "END");
-  return std::make_shared<QueryResult>(qres, pgConnection, m_connectionProvider, m_resultMapper);
+  if(!connection) {
+    throw std::runtime_error("[oatpp::postgresql::Executor::commit()]: "
+                             "Error. Can't COMMIT - NULL connection.");
+  }
+  return exec("COMMIT", connection);
 }
 
 std::shared_ptr<orm::QueryResult> Executor::rollback(const std::shared_ptr<orm::Connection>& connection) {
-  auto pgConnection = std::static_pointer_cast<postgresql::Connection>(connection);
-  PGresult *qres = PQexec(pgConnection->getHandle(), "ROLLBACK");
-  return std::make_shared<QueryResult>(qres, pgConnection, m_connectionProvider, m_resultMapper);
+  if(!connection) {
+    throw std::runtime_error("[oatpp::postgresql::Executor::commit()]: "
+                             "Error. Can't ROLLBACK - NULL connection.");
+  }
+  return exec("ROLLBACK", connection);
+}
+
+oatpp::String Executor::getSchemaVersionTableName(const oatpp::String& suffix) {
+  data::stream::BufferOutputStream stream;
+  stream << "oatpp_schema_version";
+  if (suffix && suffix->getSize() > 0) {
+    stream << "_" << suffix;
+  }
+  return stream.toString();
+}
+
+std::shared_ptr<orm::QueryResult> Executor::updateSchemaVersion(v_int64 newVersion,
+                                                                const oatpp::String& suffix,
+                                                                const std::shared_ptr<orm::Connection>& connection)
+{
+  data::stream::BufferOutputStream stream;
+  stream
+    << "UPDATE "
+    << getSchemaVersionTableName(suffix) << " "
+    << "SET version=" << newVersion << ";";
+  return exec(stream.toString(), connection, true);
 }
 
 v_int64 Executor::getSchemaVersion(const oatpp::String& suffix,
                                    const std::shared_ptr<orm::Connection>& connection)
 {
-  // TODO implement me!
+
+  std::shared_ptr<orm::QueryResult> result;
+
+  {
+    data::stream::BufferOutputStream stream;
+    stream << "CREATE TABLE IF NOT EXISTS " << getSchemaVersionTableName(suffix) << " (version BIGINT)";
+    result = exec(stream.toString(), connection);
+    if(!result->isSuccess()) {
+      throw std::runtime_error("[oatpp::postgresql::Executor::getSchemaVersion()]: "
+                               "Error. Can't create schema version table. " + result->getErrorMessage()->std_str());
+    }
+  }
+
+  data::stream::BufferOutputStream stream;
+  stream << "SELECT * FROM " << getSchemaVersionTableName(suffix);
+  result = exec(stream.toString(), result->getConnection(), true);
+  if(!result->isSuccess()) {
+    throw std::runtime_error("[oatpp::postgresql::Executor::getSchemaVersion()]: "
+                             "Error. Can't get schema version. " + result->getErrorMessage()->std_str());
+  }
+
+  auto rows = result->fetch<oatpp::Vector<oatpp::Object<VersionRow>>>();
+
+  if(rows->size() == 0) {
+
+    stream.setCurrentPosition(0);
+    stream << "INSERT INTO " << getSchemaVersionTableName(suffix) << " (version) VALUES (0)";
+    result = exec(stream.toString(), result->getConnection(), true);
+
+    if(result->isSuccess()) {
+      return 0;
+    }
+
+    throw std::runtime_error("[oatpp::postgresql::Executor::getSchemaVersion()]: "
+                             "Error. Can't init schema version. " + result->getErrorMessage()->std_str());
+
+  } else if(rows->size() == 1) {
+
+    auto row = rows[0];
+    if(!row->version) {
+      throw std::runtime_error("[oatpp::postgresql::Executor::getSchemaVersion()]: "
+                               "Error. The schema version table is corrupted - version is null.");
+    }
+
+    return row->version;
+
+  }
+
+  throw std::runtime_error("[oatpp::postgresql::Executor::getSchemaVersion()]: "
+                           "Error. The schema version table is corrupted - multiple version rows.");
+
 }
 
 void Executor::migrateSchema(const oatpp::String& script,
                              v_int64 newVersion,
                              const oatpp::String& suffix,
                              const std::shared_ptr<orm::Connection>& connection)
 {
-  // TODO implement me!
+
+  if(!script) {
+    throw std::runtime_error("[oatpp::postgresql::Executor::migrateSchema()]: Error. Script is null.");
+  }
+
+  if(!connection) {
+    throw std::runtime_error("[oatpp::postgresql::Executor::migrateSchema()]: Error. Connection is null.");
+  }
+
+  auto currVersion = getSchemaVersion(suffix, connection);
+  if(newVersion <= currVersion) {
+    return;
+  }
+
+  if(newVersion > currVersion + 1) {
+    throw std::runtime_error("[oatpp::postgresql::Executor::migrateSchema()]: Error. +1 version increment is allowed only.");
+  }
+
+  if(script->getSize() == 0) {
+    OATPP_LOGW("[oatpp::postgresql::Executor::migrateSchema()]", "Warning. Executing empty script for version %d", newVersion);
+  }
+
+  {
+
+    orm::Transaction transaction(this, connection);
+
+    std::shared_ptr<orm::QueryResult> result;
+
+    result = exec(script, connection);
+    if(!result->isSuccess()) {
+      OATPP_LOGE("[oatpp::postgresql::Executor::migrateSchema()]",
+                 "Error. Migration failed for version %d. %s", newVersion, result->getErrorMessage()->c_str());
+      throw std::runtime_error("[oatpp::postgresql::Executor::migrateSchema()]: "
+                               "Error. Migration failed. " + result->getErrorMessage()->std_str());
+
+    }
+
+    result = updateSchemaVersion(newVersion, suffix, connection);
+
+    if(!result->isSuccess() || result->hasMoreToFetch() > 0) {
+      throw std::runtime_error("[oatpp::postgresql::Executor::migrateSchema()]: Error. Migration failed. Can't set new version.");
+    }
+
+    result = transaction.commit();
+    if(!result->isSuccess()) {
+      throw std::runtime_error("[oatpp::postgresql::Executor::migrateSchema()]: Error. Migration failed. Can't commit.");
+    }
+
+  }
+
 }
 
 }}

src/oatpp-postgresql/Executor.hpp

@@ -65,6 +65,19 @@ class Executor : public orm::Executor {
   };
 
 private:
+
+  std::shared_ptr<orm::QueryResult> exec(const oatpp::String& statement,
+                                         const std::shared_ptr<orm::Connection>& connection,
+                                         bool useExecParams = false);
+
+  oatpp::String getSchemaVersionTableName(const oatpp::String& suffix);
+
+  std::shared_ptr<orm::QueryResult> updateSchemaVersion(v_int64 newVersion,
+                                                        const oatpp::String& suffix,
+                                                        const std::shared_ptr<orm::Connection>& connection);
+
+private:
+
   std::unique_ptr<Oid[]> getParamTypes(const StringTemplate& queryTemplate, const ParamsTypeMap& paramsTypeMap);
 
   std::shared_ptr<QueryResult> prepareQuery(const StringTemplate& queryTemplate,

src/oatpp-postgresql/orm.hpp

@@ -0,0 +1,35 @@
+/***************************************************************************
+ *
+ * Project         _____    __   ____   _      _
+ *                (  _  )  /__\ (_  _)_| |_  _| |_
+ *                 )(_)(  /(__)\  )( (_   _)(_   _)
+ *                (_____)(__)(__)(__)  |_|    |_|
+ *
+ *
+ * Copyright 2018-present, Leonid Stryzhevskyi <lganzzzo@gmail.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ ***************************************************************************/
+
+#ifndef oatpp_postgresql_orm_hpp
+#define oatpp_postgresql_orm_hpp
+
+#include "Executor.hpp"
+#include "Types.hpp"
+
+#include "oatpp/orm/SchemaMigration.hpp"
+#include "oatpp/orm/DbClient.hpp"
+#include "oatpp/core/macro/codegen.hpp"
+
+#endif // oatpp_postgresql_orm_hpp

test/CMakeLists.txt

@@ -1,6 +1,15 @@
+add_definitions (
+        -DTEST_DB_URL="postgresql://postgres:db-pass@localhost:5432/postgres"
+        -DTEST_DB_MIGRATION="${CMAKE_CURRENT_SOURCE_DIR}/oatpp-postgresql/migration/"
+)
+
 add_executable(module-tests
+        oatpp-postgresql/types/IntTest.cpp
+        oatpp-postgresql/types/IntTest.hpp
+        oatpp-postgresql/types/FloatTest.cpp
+        oatpp-postgresql/types/FloatTest.hpp
         oatpp-postgresql/tests.cpp
-)
+        )
 
 set_target_properties(module-tests PROPERTIES
         CXX_STANDARD 11

test/oatpp-postgresql/migration/FloatTest.sql

@@ -0,0 +1,21 @@
+DROP TABLE IF EXISTS test_floats;
+
+CREATE TABLE test_floats (
+  f_decimal         decimal,
+  f_number          numeric(1000),
+  f_real            real,
+  f_double          double precision
+);
+
+INSERT INTO test_floats
+(f_decimal, f_number, f_real, f_double) VALUES (null, null, null, null);
+
+INSERT INTO test_floats
+(f_decimal, f_number, f_real, f_double) VALUES (0, 0, 0, 0);
+
+INSERT INTO test_floats
+(f_decimal, f_number, f_real, f_double) VALUES (0.1, 0.1, 0.1, 0.1);
+
+INSERT INTO test_floats
+(f_decimal, f_number, f_real, f_double) VALUES (-0.1, -0.1, -0.1, -0.1);
+

test/oatpp-postgresql/migration/IntTest.sql

@@ -0,0 +1,7 @@
+DROP TABLE IF EXISTS test_ints;
+
+CREATE TABLE test_ints (
+  f_int16   smallint,
+  f_int32   integer,
+  f_int64   bigint
+);