Repository: elastic-rs/elastic (language: rust)

Issue:
bulk insert operation wraps every document in 'doc'
hi, I have an issue, I am trying to use bulk insert and my document's body is always wrapped by a Doc struct
![image](https://user-images.githubusercontent.com/8044785/56328000-81819600-6142-11e9-823c-4b1ce314ffe0.png)

According to the official documentation, bulk insert operation must not provide a 'doc' wrapper, I's only necessary for updating operations.
https://www.elastic.co/guide/en/elasticsearch/reference/6.2/docs-bulk.html
{ "index" : { "_index" : "test", "_type" : "_doc", "_id" : "1" } }
{ "field1" : "value1" } 



Discussion / hints:
Hi @bauto17 :wave:

Could you share which version of the client you're using? I believe we [fixed this up recently](https://github.com/elastic-rs/elastic/commit/ad941241f781a13f35c336ddd2d495b87b6247a8), so you might just need to update to the current `0.21.0-pre.3` release.
It's still in use in BulkRawOperation struct in this path src/elastic/src/client/requests/bulk/operation.rs

Fix patch:
diff --git a/src/elastic/src/client/requests/bulk/operation.rs b/src/elastic/src/client/requests/bulk/operation.rs
index 1ca2be3493..024196530f 100644
--- a/src/elastic/src/client/requests/bulk/operation.rs
+++ b/src/elastic/src/client/requests/bulk/operation.rs
@@ -300,7 +300,7 @@ impl BulkRawOperation {
         BulkRawOperation { _private: () }
     }
 
-    pub fn index<TDocument>(self, doc: TDocument) -> BulkOperation<Doc<TDocument>> {
+    pub fn index<TDocument>(self, doc: TDocument) -> BulkOperation<TDocument> {
         BulkOperation {
             action: Action::Index,
             header: BulkHeader {
@@ -308,7 +308,7 @@ impl BulkRawOperation {
                 ty: None,
                 id: None,
             },
-            inner: Some(Doc::value(doc)),
+            inner: Some(doc),
         }
     }
 
@@ -360,7 +360,7 @@ impl BulkRawOperation {
         .script_fluent(builder)
     }
 
-    pub fn create<TDocument>(self, doc: TDocument) -> BulkOperation<Doc<TDocument>> {
+    pub fn create<TDocument>(self, doc: TDocument) -> BulkOperation<TDocument> {
         BulkOperation {
             action: Action::Create,
             header: BulkHeader {
@@ -368,7 +368,7 @@ impl BulkRawOperation {
                 ty: None,
                 id: None,
             },
-            inner: Some(Doc::value(doc)),
+            inner: Some(doc),
         }
     }
 


Test patch:
diff --git a/tests/run/src/bulk/index_get.rs b/tests/run/src/bulk/index_get.rs
index f8a3b20ed4..7e0659b4fa 100644
--- a/tests/run/src/bulk/index_get.rs
+++ b/tests/run/src/bulk/index_get.rs
@@ -7,7 +7,7 @@ use run_tests::IntegrationTest;
 pub struct IndexGet;
 
 #[derive(Debug, PartialEq, Serialize, Deserialize, ElasticType)]
-#[elastic(ty = "document", index = "bulk_index")]
+#[elastic(ty = "document", index = "bulk_index_get")]
 pub struct Doc {
     #[elastic(id)]
     id: String,
diff --git a/tests/run/src/bulk/mod.rs b/tests/run/src/bulk/mod.rs
index 8ef20af88f..034787f445 100644
--- a/tests/run/src/bulk/mod.rs
+++ b/tests/run/src/bulk/mod.rs
@@ -6,6 +6,8 @@ use run_tests::{
 mod delete;
 mod index_create;
 mod index_get;
+mod raw_index_create;
+mod raw_index_get;
 mod stream;
 mod stream_tiny_size_limit;
 mod stream_tiny_timeout;
@@ -16,6 +18,8 @@ pub fn tests() -> Vec<Test> {
         Box::new(|client| test(client, delete::Delete)),
         Box::new(|client| test(client, index_get::IndexGet)),
         Box::new(|client| test(client, index_create::IndexCreate)),
+        Box::new(|client| test(client, raw_index_get::RawIndexGet)),
+        Box::new(|client| test(client, raw_index_create::RawIndexCreate)),
         Box::new(|client| test(client, stream::BulkStream)),
         Box::new(|client| test(client, stream_tiny_size_limit::BulkStreamTinySize)),
         Box::new(|client| test(client, stream_zero_size_limit::BulkStreamZeroSize)),
diff --git a/tests/run/src/bulk/raw_index_create.rs b/tests/run/src/bulk/raw_index_create.rs
new file mode 100644
index 0000000000..68807a0ece
--- /dev/null
+++ b/tests/run/src/bulk/raw_index_create.rs
@@ -0,0 +1,63 @@
+use elastic::error::Error;
+use elastic::prelude::*;
+use futures::Future;
+use run_tests::IntegrationTest;
+
+#[derive(Debug, Clone, Copy)]
+pub struct RawIndexCreate;
+
+#[derive(Debug, PartialEq, Serialize, Deserialize, ElasticType)]
+pub struct Doc {
+    #[elastic(id)]
+    id: String,
+    title: String,
+    timestamp: Date<DefaultDateMapping>,
+}
+
+const INDEX: &'static str = "raw_bulk_index_create";
+const ID: &'static str = "1";
+
+fn doc() -> Doc {
+    Doc {
+        id: ID.to_owned(),
+        title: "A document title".to_owned(),
+        timestamp: Date::build(2017, 03, 24, 13, 44, 0, 0),
+    }
+}
+
+impl IntegrationTest for RawIndexCreate {
+    type Response = BulkResponse;
+
+    fn kind() -> &'static str {
+        "bulk"
+    }
+    fn name() -> &'static str {
+        "index create"
+    }
+
+    // Ensure the index doesn't exist
+    fn prepare(&self, client: AsyncClient) -> Box<Future<Item = (), Error = Error>> {
+        let delete_res = client.index(INDEX).delete().send().map(|_| ());
+
+        Box::new(delete_res)
+    }
+
+    // Index a document, then get it
+    fn request(&self, client: AsyncClient) -> Box<Future<Item = Self::Response, Error = Error>> {
+        let bulk_res = client.bulk().push(bulk_raw()
+            .create(doc())
+            .index(INDEX)
+            .ty(Doc::static_ty())
+            .id(ID))
+            .send();
+
+        Box::new(bulk_res)
+    }
+
+    // Ensure the response contains the expected document
+    fn assert_ok(&self, res: &Self::Response) -> bool {
+        let created = res.iter().next().unwrap().unwrap();
+
+        created.action() == BulkAction::Create && created.created()
+    }
+}
diff --git a/tests/run/src/bulk/raw_index_get.rs b/tests/run/src/bulk/raw_index_get.rs
new file mode 100644
index 0000000000..056b5c9659
--- /dev/null
+++ b/tests/run/src/bulk/raw_index_get.rs
@@ -0,0 +1,68 @@
+use elastic::error::Error;
+use elastic::prelude::*;
+use futures::Future;
+use run_tests::IntegrationTest;
+
+#[derive(Debug, Clone, Copy)]
+pub struct RawIndexGet;
+
+#[derive(Debug, PartialEq, Serialize, Deserialize, ElasticType)]
+#[elastic(ty = "document", index = "raw_bulk_index_get")]
+pub struct Doc {
+    #[elastic(id)]
+    id: String,
+    title: String,
+    timestamp: Date<DefaultDateMapping>,
+}
+
+impl IntegrationTest for RawIndexGet {
+    type Response = GetResponse<Doc>;
+
+    fn kind() -> &'static str {
+        "bulk"
+    }
+    fn name() -> &'static str {
+        "index"
+    }
+
+    // Ensure the index doesn't exist
+    fn prepare(&self, client: AsyncClient) -> Box<Future<Item = (), Error = Error>> {
+        let delete_res = client
+            .index(Doc::static_index())
+            .delete()
+            .send()
+            .map(|_| ());
+
+        Box::new(delete_res)
+    }
+
+    // Index some bulk documents
+    fn request(&self, client: AsyncClient) -> Box<Future<Item = Self::Response, Error = Error>> {
+        let ops = (0..10).into_iter().map(|i| {
+            bulk_raw()
+                .index(Doc {
+                    id: i.to_string(),
+                    title: "A document title".to_owned(),
+                    timestamp: Date::build(2017, 03, 24, 13, 44, 0, 0),
+                })
+                .index(Doc::static_index())
+                .ty(Doc::static_ty())
+                .id(i.to_string())
+        });
+
+        let bulk_res = client
+            .bulk()
+            .extend(ops)
+            .params_fluent(|p| p.url_param("refresh", true))
+            .send();
+
+        let get_res = client.document().get("4").send();
+
+        Box::new(bulk_res.and_then(|_| get_res))
+    }
+
+    // Ensure the response contains the expected document
+    fn assert_ok(&self, res: &Self::Response) -> bool {
+        res.document().map(|doc| &*doc.id) == Some("4")
+    }
+}