Merge branch 'wl10981-single-crud'

# Conflicts:
#	cdk/include/mysql/cdk/protocol/mysqlx.h
#	cdk/mysqlx/session.cc
#	devapi/tests/crud-t.cc
#	xapi/tests/xapi_crud-t.cc

Author: rsomla1

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/collection_crud.cc

@@ -85,13 +85,15 @@ class Op_collection_add
   std::vector<mysqlx::GUID> m_id_list;
   bool  m_generated_id;
   unsigned m_pos;
+  bool m_upsert = false;
 
 
-  Op_collection_add(Collection &coll)
+  Op_collection_add(Collection &coll, bool upsert = false)
     : Op_base(coll)
     , m_coll(coll)
     , m_generated_id(true)
     , m_pos(0)
+    , m_upsert(upsert)
   {}
 
   Executable_impl* clone() const override
@@ -116,7 +118,8 @@ class Op_collection_add
     // Issue coll_add statement where documents are described by list
     // of expressions defined by this instance.
 
-    return new cdk::Reply(get_cdk_session().coll_add(m_coll, *this, NULL));
+    return new cdk::Reply(get_cdk_session()
+                          .coll_add(m_coll, *this, NULL, m_upsert));
   }
 
 
@@ -214,6 +217,7 @@ class Op_collection_add
 
 
   friend mysqlx::CollectionAdd;
+  friend mysqlx::internal::CollectionReplace;
 };
 
 
@@ -546,6 +550,14 @@ class Op_collection_modify
     add_where(expr);
   }
 
+  Op_collection_modify(const Op_collection_modify &other)
+    : Op_select(other)
+    , m_coll(other.m_coll)
+    , m_update(other.m_update)
+  {
+    m_update_it = m_update.end();
+  }
+
   Executable_impl* clone() const override
   {
     return new Op_collection_modify(*this);
@@ -640,6 +652,7 @@ class Op_collection_modify
   }
 
   friend mysqlx::CollectionModify;
+  friend mysqlx::internal::CollectionReplace;
 };
 
 
@@ -653,3 +666,50 @@ CollectionModify::CollectionModify(
   }
   CATCH_AND_WRAP
 }
+
+internal::CollectionReplace::CollectionReplace(Collection &coll,
+                                               const mysqlx::string &id,
+                                               mysqlx::internal::ExprValue &&val,
+                                               bool upsert)
+{
+  try
+  {
+    if (upsert)
+    {
+      Op_collection_add *add_impl = new Op_collection_add(coll, upsert);
+      reset(add_impl);
+      string str = val;
+      str.erase(str.rfind('}'));
+
+      std::stringstream sstream;
+      // If ':' is present in the string the document is not empty
+      if (str.find(':'))
+        sstream << ", ";
+      // It is safe to append the _id at the end
+      sstream << "\"_id\": \"" << id << "\"}";
+      str.append((string)sstream.str());
+      add_impl->add_json(str);
+    }
+    else
+    {
+      reset(new Op_collection_modify(coll, "_id = :id"));
+
+      if (val.isExpression() || val.getType() != Value::STRING)
+      {
+        get_impl()->add_operation(Op_collection_modify::SET,
+                                  "$",
+                                  std::move(val));
+      }
+      else
+      {
+        get_impl()->add_operation(Op_collection_modify::SET,
+                                  "$",
+                                  internal::expr(std::move(val)));
+      }
+
+      get_impl()->add_param("id", id);
+    }
+  }
+  CATCH_AND_WRAP
+}
+