Added CDK support for UPSERT command

Author: Bogdan Degtyariov

cdk/core/tests/session_crud-t.cc

@@ -807,7 +807,6 @@ class Path
 
 // =====================================================================
 
-
 CRUD_TEST_BEGIN(find)
 {
   Session sess(this);
@@ -1824,3 +1823,36 @@ CRUD_TEST_BEGIN(views)
   cout <<"Done!" <<endl;
 }
 CRUD_TEST_END
+
+
+CRUD_TEST_BEGIN(upsert)
+{
+  Session sess(this);
+  SKIP_IF_SERVER_VERSION_LESS(sess, 8, 0, 3)
+
+  Doc_list doc_list;
+  doc_list.add(1, "coo", 10);
+  doc_list.add(2, "roo", 20);
+  doc_list.add(3, "moo", 30);
+
+  Reply r = sess.coll_add(coll, doc_list, &doc_list.params());
+  r.wait();
+
+  Doc_list upsert_list;
+  upsert_list.add(1, "zoo", 40);
+  r = sess.coll_add(coll, upsert_list,
+                   &upsert_list.params(), true);
+  r.wait();
+  EXPECT_EQ(0, r.entry_count());
+
+  Doc_list no_upsert_list;
+  no_upsert_list.add(1, "noo", 50);
+  r = sess.coll_add(coll, no_upsert_list,
+                    &no_upsert_list.params(), false);
+  r.wait();
+  EXPECT_EQ(1, r.entry_count());
+
+  cout << "Done!" << endl;
+
+}
+CRUD_TEST_END

cdk/core/tests/test.h

@@ -61,10 +61,88 @@ class Core_test
     Session(Core_test *fixture)
       : cdk::Session(fixture->get_ds(), fixture->get_opts())
     {}
+
+    bool is_server_version_less(int test_upper_version,
+                                int test_lower_version,
+                                int test_release_version)
+    {
+      Reply reply(sql("SELECT VERSION()"));
+      reply.wait();
+      Cursor cursor(reply);
+
+      struct : cdk::Row_processor
+      {
+        cdk::string version;
+
+        virtual bool row_begin(row_count_t row)
+        { return true; }
+
+        virtual void row_end(row_count_t row)
+        {}
+
+        virtual void field_null(col_count_t pos)
+        {}
+
+        virtual size_t field_begin(col_count_t pos, size_t)
+        { return  SIZE_MAX; }
+
+        size_t field_data(col_count_t pos, bytes data)
+        {
+          cdk::foundation::Codec<cdk::foundation::Type::STRING> codec;
+          // Trim trailing \0
+          bytes d1(data.begin(), data.end() - 1);
+          codec.from_bytes(d1, version);
+          return 0;
+        }
+
+        virtual void field_end(col_count_t /*pos*/)
+        {}
+
+        virtual void end_of_data()
+        {}
+      }
+      prc;
+
+      cursor.get_rows(prc);
+      cursor.wait();
+
+      std::stringstream version;
+      version << std::string(prc.version);
+
+      int upper_version, minor_version, release_version;
+      char sep;
+      version >> upper_version;
+      version >> sep;
+      version >> minor_version;
+      version >> sep;
+      version >> release_version;
+
+      if ((upper_version < test_upper_version) ||
+        (upper_version == test_upper_version &&
+         minor_version << test_lower_version) ||
+         (upper_version == test_upper_version &&
+          minor_version == test_lower_version &&
+          release_version < test_release_version))
+      {
+        return true;
+      }
+      return false;
+    }
+
   };
 
 };
 
+#define SKIP_IF_SERVER_VERSION_LESS(CDK_SESS, x,y,z)\
+  if (CDK_SESS.is_server_version_less(x, y, z)) \
+  {\
+    std::cerr <<"SKIPPED: " << \
+    "Server version not supported (" \
+    << x << "." << y <<"." << ")" << z <<std::endl; \
+    return; \
+  }
+
+
 template <class X>
 struct Helper
 {

cdk/include/mysql/cdk/mysqlx/session.h

@@ -522,7 +522,8 @@ class Session
 
   Reply_init &coll_add(const Table_ref&,
                        Doc_source&,
-                       const Param_source *param = NULL);
+                       const Param_source *param = NULL,
+                       bool upsert = false);
 
   Reply_init &coll_remove(const Table_ref&,
                           const Expression *expr = NULL,

cdk/include/mysql/cdk/protocol/mysqlx.h

@@ -440,7 +440,7 @@ struct Protocol_fields
     Enum values will be used as binary flags,
     so they must be as 2^N
   */
-  enum value { ROW_LOCKING = 1 /*, NEXT = 2, NEXT_NEXT = 4*/ };
+  enum value { ROW_LOCKING = 1 , UPSERT = 2 };
 };
 
 }  // api namespace
@@ -566,12 +566,19 @@ class Protocol
 
     @param args  defines values of named parameters, if any are used in the
       expressions of the row source object
+
+    @param upsert  Can be set true only in the document mode -- in that case
+      an upsert variant of the Insert command is sent. If inserted document has
+      the same id as an existing document in the collection, the upsert variant
+      replaces the document in the collection with the new one. Without upsert
+      flag such situation leads to error.
   */
 
   Op& snd_Insert(Data_model dm, api::Db_obj &obj,
                  const api::Columns *columns,
                  Row_source &data,
-                 const api::Args_map *args = NULL);
+                 const api::Args_map *args = NULL,
+                 bool upsert = false);
 
   /**
     Send CRUD Update command.

cdk/include/mysql/cdk/session.h

@@ -225,15 +225,26 @@ class Session
     Documents to be inserted are given by a Doc_source object which is
     a sequence of expressions, each expression describing a single document.
     Note that a document can be represented as a JSON blob or as a structured
-    document expression.
+    document expression. In the latter case this expression can contain named
+    parameters -- the values for these parameters are given by the `param`
+    argument describing a key-value dictionary.
+
+    If `upsert` flag is set and a document being added has the same id as
+    a document already present in the collection, the existing document is
+    replaced by the new one. Otherwise, if `upsert` flag is false (the default)
+    an error is reported if a document being added conflicts with an exisiting
+    document in the collection.
 
     Note: Server requires that inserted documents contain "_id" field with
     unique document id.
   */
 
-  Reply_init coll_add(const api::Object_ref &coll, Doc_source &docs, const Param_source *param)
+  Reply_init coll_add(const api::Object_ref &coll,
+                      Doc_source &docs,
+                      const Param_source *param,
+                      bool upsert = false)
   {
-    return m_session->coll_add(coll, docs, param);
+    return m_session->coll_add(coll, docs, param, upsert);
   }
 
   /**

cdk/mysqlx/delayed_op.h

@@ -189,6 +189,7 @@ class SndInsertDocs
 
   cdk::Doc_source &m_docs;
   const Param_source *m_param;
+  bool m_upsert;
 
   Proto_op* start()
   {
@@ -201,17 +202,20 @@ class SndInsertDocs
                                   *this,
                                   NULL,
                                   *this,
-                                  &param_conv);
+                                  &param_conv,
+                                  m_upsert);
   }
 
 public:
 
   SndInsertDocs(Protocol& protocol, const api::Table_ref &coll,
                 cdk::Doc_source &docs,
-                const Param_source *param)
+                const Param_source *param,
+                bool upsert = false)
     : Crud_op_base(protocol, coll)
     , m_docs(docs)
     , m_param(param)
+    , m_upsert(upsert)
   {}
 
 private:

cdk/mysqlx/session.cc

@@ -84,6 +84,10 @@ struct Proto_field_checker : public cdk::protocol::mysqlx::api::Expectations
         // Find(17) locking(12)
         m_data = bytes("17.12");
         break;
+      case Protocol_fields::UPSERT:
+        // Insert(18) upsert(6)
+        m_data = bytes("18.6");
+        break;
       default:
         return 0;
     }
@@ -373,6 +377,7 @@ void Session::check_protocol_fields()
     m_proto_fields = 0;
     /* More fields checks will be added here */
     m_proto_fields |= field_checker.is_supported(Protocol_fields::ROW_LOCKING);
+    m_proto_fields |= field_checker.is_supported(Protocol_fields::UPSERT);
   }
 }
 
@@ -467,10 +472,11 @@ void Session::rollback()
 
 Reply_init& Session::coll_add(const Table_ref &coll,
                               Doc_source &docs,
-                              const Param_source *param)
+                              const Param_source *param,
+                              bool upsert)
 {
   return set_command(
-    new SndInsertDocs(m_protocol, coll, docs, param)
+    new SndInsertDocs(m_protocol, coll, docs, param, upsert)
   );
 }
 

cdk/protocol/mysqlx/crud.cc

@@ -555,7 +555,8 @@ Protocol::snd_Insert(
     api::Db_obj &db_obj,
     const api::Columns *columns,
     Row_source &rs,
-    const api::Args_map *args)
+    const api::Args_map *args,
+    bool upsert)
 {
   Mysqlx::Crud::Insert insert;
 
@@ -584,6 +585,8 @@ Protocol::snd_Insert(
     rs.process(row_builder);
   }
 
+  insert.set_upsert(upsert);
+
   return get_impl().snd_start(insert, msg_type::cli_CrudInsert);
 }
 

devapi/tests/crud-t.cc

@@ -2209,6 +2209,8 @@ TEST_F(Crud, single_document)
 {
   SKIP_IF_NO_XPLUGIN;
 
+  SKIP_IF_SERVER_VERSION_LESS(8, 0, 2);
+
   cout << "Creating session..." << endl;
 
   Session sess(this);