pgsql does not support INSERT ... ON DUPLICATE KEY UPDATE, switch to …

…UPDATE ... or if no rows affected then INSERT

git-svn-id: http://kazuho.31tools.com/svn/incline/trunk@248 4d3e2a30-9d6d-0410-bc8c-dac56cff10b3

src/incline_dbms.h

@@ -62,7 +62,7 @@ class incline_dbms {
   std::string host() const { return host_; }
   unsigned short port() const { return port_; }
   virtual std::string escape(const std::string& s) = 0;
-  virtual void execute(const std::string& stmt) = 0;
+  virtual unsigned long long execute(const std::string& stmt) = 0;
   void query(std::vector<std::vector<value_t> >& rows, const char* fmt, ...) __attribute__((__format__(__printf__, 3, 4)));
   virtual void query(std::vector<std::vector<value_t> >& rows, const std::string& stmt) = 0;
   virtual std::string get_column_def(const std::string& table_name, const std::string& column_name) = 0;

src/incline_fw_replicator.cc

@@ -87,9 +87,15 @@ incline_fw_replicator::run()
       if (! insert_rows.empty()) {
 	this->insert_rows(dest_dbh, insert_rows);
       }
-      string sql("INSERT INTO _iq_repl (tbl_name,last_id) VALUES ('" + def()->destination() + "'," + last_id + ") ON DUPLICATE KEY UPDATE last_id=VALUES(last_id)");
+      string sql("UPDATE _iq_repl SET last_id=" + last_id + " WHERE tbl_name='"
+		 + dest_dbh->escape(def()->destination()) + '\'');
       mgr_->log_sql(dest_dbh, sql);
-      dest_dbh->execute(sql);
+      if (dest_dbh->execute(sql) == 0) {
+	sql = "INSERT INTO _iq_repl (tbl_name,last_id) VALUES ('"
+	  + dest_dbh->escape(def()->destination()) + "'," + last_id + ')';
+	mgr_->log_sql(dest_dbh, sql);
+	dest_dbh->execute(sql);
+      }
       dest_dbh->execute("COMMIT");
     } catch (incline_dbms::error_t& err) {
       cerr << err.what() << endl;

src/incline_mysql.cc

@@ -70,7 +70,7 @@ incline_mysql::escape(const string& s)
   }
 }
 
-void
+unsigned long long
 incline_mysql::execute(const string& stmt)
 {
   try {
@@ -85,6 +85,7 @@ incline_mysql::execute(const string& stmt)
       throw error_t(e.what());
     }
   }
+  return tmd::affected_rows(*dbh_);
 }
 
 void

src/incline_mysql.h

@@ -27,7 +27,7 @@ class incline_mysql : public incline_dbms {
 public:
   virtual ~incline_mysql();
   virtual std::string escape(const std::string& s);
-  virtual void execute(const std::string& stmt);
+  virtual unsigned long long execute(const std::string& stmt);
   virtual void query(std::vector<std::vector<value_t> >& rows, const std::string& stmt);
   virtual std::string get_column_def(const std::string& table_name, const std::string& column_name);
 private:

src/incline_pgsql.cc

@@ -102,7 +102,7 @@ incline_pgsql::escape(const string& s)
   return string(buf);
 }
 
-void
+unsigned long long
 incline_pgsql::execute(const string& stmt)
 {
   PGresultWrap ret(PQexec(dbh_, stmt.c_str()));
@@ -118,6 +118,13 @@ incline_pgsql::execute(const string& stmt)
   default:
     THROW_PQ_ERROR(dbh_);
   }
+  unsigned long long nrows = 0;
+  if (const char* s = PQcmdTuples(*ret)) {
+    if (s[0] != '\0') {
+      nrows = strtoll(s, NULL, 10);
+    }
+  }
+  return nrows;
 }
 
 void

src/incline_pgsql.h

@@ -23,7 +23,7 @@ class incline_pgsql : public incline_dbms {
 public:
   virtual ~incline_pgsql();
   virtual std::string escape(const std::string& s);
-  virtual void execute(const std::string& stmt);
+  virtual unsigned long long execute(const std::string& stmt);
   virtual void query(std::vector<std::vector<value_t> >& rows, const std::string& stmt);
  virtual std::string get_column_def(const std::string& table_name, const std::string& column_name);
 private: