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

Issue:
Ability to return `_source` from an update bulk request
Currently, I don't believe you can request to get the `_source` back using the `BulkRequestBuilder` API like [this](https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html#bulk-update)?

It would be really helpful to expose that option...

I looked into implementing this - it seems like I would need to change things in `elastic/src/elastic/src/client/requests/bulk/*`. Is that correct? Anywhere else?

The catch is that I believe only the `update` action supports returning the `_source`, so we would need to figure out how to best expose this... Maybe we add an extra parameter to the [`update` method](https://github.com/elastic-rs/elastic/blob/f16a3720dba6a940c804650e463b3cd25c614a88/src/elastic/src/client/requests/bulk/operation.rs#L204)?


Discussion / hints:

Fix patch:
diff --git a/src/elastic/examples/bulk_with_source.rs b/src/elastic/examples/bulk_with_source.rs
new file mode 100644
index 0000000000..ac29474b49
--- /dev/null
+++ b/src/elastic/examples/bulk_with_source.rs
@@ -0,0 +1,74 @@
+//! Send a bulk request and return the updated document at the same time.
+//!
+//! NOTE: This sample expects you have a node running on `localhost:9200`.
+//!
+//! This sample demonstrates a bulk request.
+//! There are a few extra dials you can tweak on bulk requests to get more
+//! performance out of them.
+//! See the docs for `BulkResponse` for more details.
+
+use elastic::prelude::*;
+use elastic_derive::ElasticType;
+use serde::{
+    Deserialize,
+    Serialize,
+};
+use serde_json::json;
+
+#[derive(Serialize, Deserialize, ElasticType)]
+#[elastic(index = "bulk_with_source_sample_index")]
+struct NewsArticle {
+    id: i64,
+    likes: i64,
+}
+#[derive(Serialize, Deserialize, ElasticType, Debug)]
+#[elastic(index = "bulk_with_source_sample_index")]
+struct UpdatedNewsArticle {
+    id: i64,
+    likes: i64,
+}
+
+fn run() -> Result<(), Box<dyn::std::error::Error>> {
+    let client = SyncClientBuilder::new().build()?;
+
+    let index_ops = (0..100)
+        .into_iter()
+        .map(|i| bulk_raw().index(json!({ "id": i, "likes": 0 })).id(i));
+
+    // Index some documents
+    let _response = client
+        .bulk()
+        .index(NewsArticle::static_index())
+        .ty(NewsArticle::static_ty())
+        .extend(index_ops)
+        .params_fluent(|p| p.url_param("refresh", true))
+        .send()?;
+
+    let update_ops = (0..100).into_iter().map(|i| {
+        bulk::<NewsArticle>()
+            .update_script(i, "ctx._source.likes++")
+            // Request that the updated document's source be returned with the response
+            .source(true)
+    });
+
+    // Update the documents
+    let response = client
+        .bulk()
+        .index(NewsArticle::static_index())
+        .ty(NewsArticle::static_ty())
+        .extend(update_ops)
+        .send()?;
+
+    for op in response {
+        if let Ok(op) = op {
+            println!("{:?}", op.into_document::<UpdatedNewsArticle>().unwrap());
+        }
+    }
+
+    Ok(())
+}
+
+fn main() {
+    env_logger::init();
+    run().unwrap()
+}
diff --git a/src/elastic/src/client/requests/bulk/operation.rs b/src/elastic/src/client/requests/bulk/operation.rs
index 2e9bdf84c9..787af6b93e 100644
--- a/src/elastic/src/client/requests/bulk/operation.rs
+++ b/src/elastic/src/client/requests/bulk/operation.rs
@@ -12,7 +12,7 @@ use serde::ser::{
     SerializeMap,
     Serializer,
 };
-use serde_json;
+use serde_json::Value;
 
 use crate::{
     client::requests::common::{
@@ -91,6 +91,64 @@ impl<TParams> BulkOperation<Script<TParams>> {
             inner,
         }
     }
+
+    /**
+    Control the [source filtering] for this operation in order to return part or
+    all of the updated document in the response.
+
+    # Examples
+
+    To obtain the updated document, use `source` with the [`into_document`]
+    method of the [`OkItem`] response:
+
+    ```no_run
+    # #[macro_use] extern crate serde_derive;
+    # #[macro_use] extern crate elastic_derive;
+    # use elastic::prelude::*;
+    # fn main() { run().unwrap() }
+    # fn run() -> Result<(), Box<dyn ::std::error::Error>> {
+    # #[derive(Serialize, Deserialize, ElasticType, Debug)]
+    # struct NewsArticle { id: i64, likes: i64 }
+    # #[derive(Serialize, Deserialize, ElasticType, Debug)]
+    # struct UpdatedNewsArticle { id: i64, likes: i64 }
+    # let client = SyncClientBuilder::new().build()?;
+    let update_ops = (0..10).into_iter().map(|i| {
+        bulk::<NewsArticle>()
+            .update_script(i, "ctx._source.likes++")
+            .source(true)
+    });
+
+    let response = client
+        .bulk()
+        .index(NewsArticle::static_index())
+        .ty(NewsArticle::static_ty())
+        .extend(update_ops)
+        .send()?;
+
+    for op in response {
+        if let Ok(op) = op {
+            println!("{:?}", op.into_document::<UpdatedNewsArticle>().unwrap());
+        }
+    }
+
+    # Ok(())
+    # }
+    ```
+
+    [source filtering]: https://www.elastic.co/guide/en/elasticsearch/reference/current/search-request-body.html#request-body-search-source-filtering
+    [`into_document`]: ../../responses/bulk/struct.OkItem.html#method.into_document
+    [`OkItem`]: ../../responses/bulk/struct.OkItem.html
+    */
+    pub fn source<T>(mut self, value: T) -> Self
+    where
+        T: Into<Value>,
+    {
+        if let Some(inner) = self.inner {
+            self.inner = Some(inner.source(value));
+        };
+
+        self
+    }
 }
 
 impl<TValue> BulkOperation<TValue> {
@@ -136,6 +194,34 @@ where
 
         self
     }
+
+    /**
+    Control the [source filtering] for this operation in order to return part or
+    all of the updated document in the response.
+
+    To obtain the updated document, use `source` with the [`into_document`]
+    method of the [`OkItem`] response.
+
+    # Examples
+
+    See the [`source`] method on [`BulkOperation<Script<TParams>>`].
+
+    [source filtering]: https://www.elastic.co/guide/en/elasticsearch/reference/current/search-request-body.html#request-body-search-source-filtering
+    [`into_document`]: ../../responses/bulk/struct.OkItem.html#method.into_document
+    [`OkItem`]: ../../responses/bulk/struct.OkItem.html
+    [`source`]: struct.BulkOperation.html#method.source
+    [`BulkOperation<Script<TParams>>`]: struct.BulkOperation.html
+    */
+    pub fn source<T>(mut self, value: T) -> Self
+    where
+        T: Into<Value>,
+    {
+        if let Some(inner) = self.inner {
+            self.inner = Some(inner.source(value));
+        };
+
+        self
+    }
 }
 
 impl<TDocument> BulkOperation<TDocument>
diff --git a/src/elastic/src/client/requests/common.rs b/src/elastic/src/client/requests/common.rs
index 2a22dc04de..c463165869 100644
--- a/src/elastic/src/client/requests/common.rs
+++ b/src/elastic/src/client/requests/common.rs
@@ -19,6 +19,8 @@ pub struct Doc<TDocument> {
     doc: DocInner<TDocument>,
     #[serde(skip_serializing_if = "Not::not")]
     doc_as_upsert: bool,
+    #[serde(rename = "_source", skip_serializing_if = "Option::is_none")]
+    source: Option<Value>,
 }
 
 impl<TDocument> Doc<TDocument> {
@@ -26,6 +28,7 @@ impl<TDocument> Doc<TDocument> {
         Doc {
             doc: DocInner { inner: None },
             doc_as_upsert: false,
+            source: None,
         }
     }
 
@@ -33,6 +36,7 @@ impl<TDocument> Doc<TDocument> {
         Doc {
             doc: DocInner { inner: Some(doc) },
             doc_as_upsert: false,
+            source: None,
         }
     }
 
@@ -41,6 +45,13 @@ impl<TDocument> Doc<TDocument> {
 
         self
     }
+
+    /** Modify this `Doc` by adding a `source` field. */
+    pub(crate) fn source(mut self, value: impl Into<Value>) -> Self {
+        self.source = Some(value.into());
+
+        self
+    }
 }
 
 struct DocInner<TDocument> {
@@ -69,6 +80,17 @@ pub type DefaultParams = Map<String, Value>;
 #[derive(Serialize)]
 pub struct Script<TParams> {
     script: ScriptInner<TParams>,
+    #[serde(rename = "_source", skip_serializing_if = "Option::is_none")]
+    source: Option<Value>,
+}
+
+impl<TParams> Script<TParams> {
+    /** Modify this `Script` by adding a `source` field. */
+    pub(crate) fn source(mut self, value: impl Into<Value>) -> Self {
+        self.source = Some(value.into());
+
+        self
+    }
 }
 
 impl Script<DefaultParams> {
@@ -161,6 +183,7 @@ impl<TParams> ScriptBuilder<TParams> {
                 params: self.params,
                 lang: self.lang,
             },
+            source: None,
         }
     }
 }
diff --git a/src/elastic/src/client/responses/bulk.rs b/src/elastic/src/client/responses/bulk.rs
index d3b8769bda..89dec56e8e 100644
--- a/src/elastic/src/client/responses/bulk.rs
+++ b/src/elastic/src/client/responses/bulk.rs
@@ -9,6 +9,7 @@ use super::common::{
 };
 use serde::de::{
     Deserialize,
+    DeserializeOwned,
     Deserializer,
     Error as DeError,
     MapAccess,
@@ -427,6 +428,7 @@ pub struct OkItem<
     primary_term: Option<u32>,
     shards: Option<Shards>,
     result: Option<DocumentResult>,
+    get: Option<Value>,
 }
 
 impl<TIndex, TType, TId> OkItem<TIndex, TType, TId> {
@@ -496,6 +498,70 @@ impl<TIndex, TType, TId> OkItem<TIndex, TType, TId> {
     pub fn id(&self) -> &TId {
         &self.id
     }
+
+    /**
+    Convert the source in the response into the updated document.
+
+    The [`source`] method must have been called first on the [`BulkOperation`],
+    otherwise this will return `None`.
+
+    Although not a requirement, be careful that both the document and
+    updated document types use the same index, e.g. by using the
+    [`#[elastic(index)]` attribute][index-attr].
+
+    # Examples
+
+    ```no_run
+    # #[macro_use] extern crate serde_derive;
+    # #[macro_use] extern crate elastic_derive;
+    # use elastic::prelude::*;
+    # fn main() { run().unwrap() }
+    # fn run() -> Result<(), Box<dyn ::std::error::Error>> {
+    # #[derive(Serialize, Deserialize, ElasticType, Debug)]
+    # struct NewsArticle { id: i64, likes: i64 }
+    # #[derive(Serialize, Deserialize, ElasticType, Debug)]
+    # struct UpdatedNewsArticle { id: i64, likes: i64 }
+    # let client = SyncClientBuilder::new().build()?;
+    let update_ops = (0..10).into_iter().map(|i| {
+        bulk::<NewsArticle>()
+            .update_script(i, "ctx._source.likes++")
+            .source(true)
+    });
+
+    let response = client
+        .bulk()
+        .index(NewsArticle::static_index())
+        .ty(NewsArticle::static_ty())
+        .extend(update_ops)
+        .send()?;
+
+    for op in response {
+        if let Ok(op) = op {
+            println!("{:?}", op.into_document::<UpdatedNewsArticle>().unwrap());
+        }
+    }
+
+    # Ok(())
+    # }
+    ```
+
+    [`source`]: ../../requests/bulk/struct.BulkOperation.html#method.source
+    [`BulkOperation`]: ../../requests/bulk/struct.BulkOperation.html
+    [index-attr]: ../../../types/document/index.html#specifying-a-default-index-name
+    */
+    pub fn into_document<T>(&self) -> Option<T>
+    where
+        T: DeserializeOwned,
+    {
+        self.get.as_ref().map_or_else(
+            || None,
+            |obj| {
+                obj.get("_source")
+                    .cloned()
+                    .map_or_else(|| None, |doc| serde_json::from_value::<T>(doc).ok())
+            },
+        )
+    }
 }
 
 /** A failed bulk response item. */
@@ -611,6 +677,7 @@ struct ItemDeInner<TIndex, TType, TId> {
     sequence_number: Option<u32>,
     #[serde(rename = "_primary_term")]
     primary_term: Option<u32>,
+    get: Option<Value>,
     #[serde(rename = "_shards")]
     shards: Option<Shards>,
     result: Option<DocumentResult>,
@@ -651,6 +718,7 @@ where
                 primary_term: self.inner.primary_term,
                 shards: self.inner.shards,
                 result: self.inner.result,
+                get: self.inner.get,
             })
         }
     }


Test patch: