Implement _source parameter on action and doc lines (#4339) (#4426)

russcam · web-flow · commit 9545ae6dc196 · 2020-02-24T16:59:14.000+11:00
Implement _source parameter on action and doc lines. (cherry picked from commit e4c1d35)
diff --git a/src/Nest/Aggregations/AggregateJsonConverter.cs b/src/Nest/Aggregations/AggregateJsonConverter.cs
@@ -187,8 +187,9 @@ private IAggregate GetTopHitsAggregate(JsonReader reader, JsonSerializer seriali
 			var hits = o[Parser.Hits].Children().OfType<JObject>();
 			reader.Read();
 			//using request/response serializer here because doc is wrapped in NEST's Hit<T>
-			var s = serializer.GetConnectionSettings().RequestResponseSerializer;
-			var lazyHits = hits.Select(h => new LazyDocument(h, s)).ToList();
+			var s = serializer.GetConnectionSettings().SourceSerializer;
+			var r = serializer.GetConnectionSettings().RequestResponseSerializer;
+			var lazyHits = hits.Select(h => new LazyDocument(h, s, r)).ToList();
 			return new TopHitsAggregate(lazyHits)
 			{
 				Total = total,
@@ -573,7 +574,8 @@ private IAggregate GetValueAggregate(JsonReader reader, JsonSerializer serialize
 			if (scriptedMetric != null)
 			{
 				var s = serializer.GetConnectionSettings().SourceSerializer;
-				return new ScriptedMetricAggregate(new LazyDocument(scriptedMetric, s));
+				var r = serializer.GetConnectionSettings().RequestResponseSerializer;
+				return new ScriptedMetricAggregate(new LazyDocument(scriptedMetric, s, r));
 			}
 			return valueMetric;
 		}
diff --git a/src/Nest/Aggregations/Metric/TopHits/TopHitsAggregate.cs b/src/Nest/Aggregations/Metric/TopHits/TopHitsAggregate.cs
@@ -16,7 +16,7 @@ public TopHitsAggregate() { }
 		public long Total { get; set; }
 
 		private IEnumerable<Hit<TDocument>> ConvertHits<TDocument>()
-			where TDocument : class => _hits.Select(h => h.As<Hit<TDocument>>());
+			where TDocument : class => _hits.Select(h => h.AsUsingRequestResponseSerializer<Hit<TDocument>>());
 
 		public IReadOnlyCollection<Hit<TDocument>> Hits<TDocument>()
 			where TDocument : class =>
diff --git a/src/Nest/CommonAbstractions/Fields/FieldValuesJsonConverter.cs b/src/Nest/CommonAbstractions/Fields/FieldValuesJsonConverter.cs
@@ -15,7 +15,8 @@ public override object ReadJson(JsonReader reader, Type objectType, object exist
 		{
 			var o = JObject.Load(reader);
 			var s = serializer.GetConnectionSettings().SourceSerializer;
-			var fields = o.Properties().ToDictionary(p => p.Name, p => new LazyDocument(p.Value, s));
+			var r = serializer.GetConnectionSettings().RequestResponseSerializer;
+			var fields = o.Properties().ToDictionary(p => p.Name, p => new LazyDocument(p.Value, s, r));
 			var inferrer = serializer.GetConnectionSettings().Inferrer;
 			var fieldValues = new FieldValues(inferrer, fields);
 			return fieldValues;
diff --git a/src/Nest/CommonAbstractions/LazyDocument/LazyDocument.cs b/src/Nest/CommonAbstractions/LazyDocument/LazyDocument.cs
@@ -28,12 +28,14 @@ public interface ILazyDocument
 	/// <inheritdoc />
 	public class LazyDocument : ILazyDocument
 	{
-		private readonly IElasticsearchSerializer _serializer;
+		private readonly IElasticsearchSerializer _sourceSerializer;
+		private readonly IElasticsearchSerializer _requestResponseSerializer;
 
-		internal LazyDocument(JToken token, IElasticsearchSerializer serializer)
+		internal LazyDocument(JToken token, IElasticsearchSerializer sourceSerializer, IElasticsearchSerializer requestResponseSerializer)
 		{
 			Token = token;
-			_serializer = serializer;
+			_sourceSerializer = sourceSerializer;
+			_requestResponseSerializer = requestResponseSerializer;
 		}
 
 		internal JToken Token { get; }
@@ -43,15 +45,22 @@ public T As<T>()
 		{
 			if (Token == null) return default(T);
 			using (var ms = Token.ToStream())
-				return _serializer.Deserialize<T>(ms);
+				return _sourceSerializer.Deserialize<T>(ms);
 		}
 
 		/// <inheritdoc />
 		public object As(Type objectType)
 		{
 			if (Token == null) return null;
 			using (var ms = Token.ToStream())
-				return _serializer.Deserialize(objectType, ms);
+				return _sourceSerializer.Deserialize(objectType, ms);
+		}
+
+		internal T AsUsingRequestResponseSerializer<T>()
+		{
+			if (Token == null) return default(T);
+			using (var ms = Token.ToStream())
+				return _requestResponseSerializer.Deserialize<T>(ms);
 		}
 	}
 }
diff --git a/src/Nest/CommonAbstractions/LazyDocument/LazyDocumentJsonConverter.cs b/src/Nest/CommonAbstractions/LazyDocument/LazyDocumentJsonConverter.cs
@@ -20,8 +20,9 @@ public override void WriteJson(JsonWriter writer, object value, JsonSerializer s
 		public override object ReadJson(JsonReader reader, Type objectType, object existingValue, JsonSerializer serializer)
 		{
 			var sourceSerializer = serializer.GetConnectionSettings().SourceSerializer;
+			var requestResponseSerializer = serializer.GetConnectionSettings().RequestResponseSerializer;
 			var token = reader.ReadTokenWithDateParseHandlingNone();
-			return new LazyDocument(token, sourceSerializer);
+			return new LazyDocument(token, sourceSerializer, requestResponseSerializer);
 		}
 
 		public override bool CanConvert(Type objectType) => true;
diff --git a/src/Nest/Document/Multiple/Bulk/BulkOperation/BulkUpdate.cs b/src/Nest/Document/Multiple/Bulk/BulkOperation/BulkUpdate.cs
@@ -44,6 +44,11 @@ public interface IBulkUpdateOperation<TDocument, TPartialDocument> : IBulkOperat
 		long? IfSequenceNumber { get; set; }
 
 		long? IfPrimaryTerm { get; set; }
+
+		/// <summary>
+		/// True or false to return the _source field or not, or a list of fields to return.
+		/// </summary>
+		Union<bool, ISourceFilter> Source { get; set; }
 	}
 
 	public class BulkUpdateOperation<TDocument, TPartialDocument> : BulkOperationBase, IBulkUpdateOperation<TDocument, TPartialDocument>
@@ -116,6 +121,9 @@ public BulkUpdateOperation(TDocument idFrom, TPartialDocument update, bool useId
 
 		public long? IfPrimaryTerm { get; set; }
 
+		/// <inheritdoc />
+		public Union<bool, ISourceFilter> Source { get; set; }
+
 		protected override Type ClrType => typeof(TDocument);
 
 		protected override string Operation => "update";
@@ -135,7 +143,8 @@ protected override object GetBody() =>
 				_DocAsUpsert = DocAsUpsert,
 				_ScriptedUpsert = ScriptedUpsert,
 				IfPrimaryTerm = IfPrimaryTerm,
-				IfSequenceNumber = IfSequenceNumber
+				IfSequenceNumber = IfSequenceNumber,
+				Source = Source,
 			};
 	}
 
@@ -155,6 +164,7 @@ public class BulkUpdateDescriptor<TDocument, TPartialDocument>
 		TDocument IBulkUpdateOperation<TDocument, TPartialDocument>.Upsert { get; set; }
 		long? IBulkUpdateOperation<TDocument, TPartialDocument>.IfSequenceNumber { get; set; }
 		long? IBulkUpdateOperation<TDocument, TPartialDocument>.IfPrimaryTerm { get; set; }
+		Union<bool, ISourceFilter> IBulkUpdateOperation<TDocument, TPartialDocument>.Source { get; set; }
 
 		protected override object GetBulkOperationBody() =>
 			new BulkUpdateBody<TDocument, TPartialDocument>
@@ -165,7 +175,8 @@ protected override object GetBulkOperationBody() =>
 				_DocAsUpsert = Self.DocAsUpsert,
 				_ScriptedUpsert = Self.ScriptedUpsert,
 				IfPrimaryTerm = Self.IfPrimaryTerm,
-				IfSequenceNumber = Self.IfSequenceNumber
+				IfSequenceNumber = Self.IfSequenceNumber,
+				Source = Self.Source,
 			};
 
 		protected override Id GetIdForOperation(Inferrer inferrer) =>
@@ -234,5 +245,11 @@ public BulkUpdateDescriptor<TDocument, TPartialDocument> IfSequenceNumber(long?
 		/// </summary>
 		public BulkUpdateDescriptor<TDocument, TPartialDocument> IfPrimaryTerm(long? primaryTerm) =>
 			Assign(primaryTerm, (a, v) => a.IfPrimaryTerm = v);
+
+		/// <summary>
+		/// True or false to return the _source field or not, or a list of fields to return.
+		/// </summary>
+		public BulkUpdateDescriptor<TDocument, TPartialDocument> Source(Union<bool, ISourceFilter> source) =>
+			Assign(source, (a, v) => a.Source = v);
 	}
 }
diff --git a/src/Nest/Document/Multiple/Bulk/BulkOperation/BulkUpdateBody.cs b/src/Nest/Document/Multiple/Bulk/BulkOperation/BulkUpdateBody.cs
@@ -28,5 +28,8 @@ internal class BulkUpdateBody<TDocument, TPartialUpdate>
 
 		[JsonProperty("if_primary_term")]
 		internal long? IfPrimaryTerm { get; set; }
+
+		[JsonProperty("_source")]
+		internal Union<bool, ISourceFilter> Source { get; set; }
 	}
 }
diff --git a/src/Nest/Document/Multiple/Bulk/BulkResponseItem/BulkResponseItemBase.cs b/src/Nest/Document/Multiple/Bulk/BulkResponseItem/BulkResponseItemBase.cs
@@ -72,6 +72,18 @@ public interface IBulkResponseItem
 		/// </summary>
 		[JsonProperty("_version")]
 		long Version { get; }
+
+		/// <summary>
+		/// The <see cref="Nest.GetResponse{TDocument}"/> when <see cref="BulkUpdateOperation{TDocument,TPartialDocument}.Source"/>
+		/// is <c>true</c> or specifies fields with <see cref="ISourceFilter.Includes"/>
+		/// </summary>
+		[JsonProperty("get")]
+		LazyDocument Get { get; }
+
+		/// <summary>
+		/// Deserialize the <see cref="Get"/> property as a GetResponse<TDocument> type, where TDocument is the document type.
+		/// </summary>
+		GetResponse<TDocument> GetResponse<TDocument>() where TDocument : class;
 	}
 
 	/// <inheritdoc />
@@ -130,6 +142,13 @@ public bool IsValid
 		/// <inheritdoc />
 		public long Version { get; internal set; }
 
+		/// <inheritdoc />
+		public LazyDocument Get { get; internal set; }
+
+		/// <inheritdoc />
+		public GetResponse<TDocument> GetResponse<TDocument>() where TDocument : class =>
+			Get?.AsUsingRequestResponseSerializer<GetResponse<TDocument>>();
+
 		public override string ToString() =>
 			$"{Operation} returned {Status} _index: {Index} _type: {Type} _id: {Id} _version: {Version} error: {Error}";
 	}
diff --git a/src/Nest/Document/Multiple/MultiGet/Response/MultiGetHitJsonConverter.cs b/src/Nest/Document/Multiple/MultiGet/Response/MultiGetHitJsonConverter.cs
@@ -27,10 +27,11 @@ private static void CreateMultiHit<T>(MultiHitTuple tuple, JsonSerializer serial
 			var hit = tuple.Hit.ToObject<MultiGetHit<T>>(serializer);
 			var settings = serializer.GetConnectionSettings();
 			var s = serializer.GetConnectionSettings().SourceSerializer;
+			var r = serializer.GetConnectionSettings().RequestResponseSerializer;
 
 			if (tuple.Hit["fields"] is JObject fields)
 			{
-				var fieldsDictionary = fields.Properties().ToDictionary(p => p.Name, p => new LazyDocument(p.Value, s));
+				var fieldsDictionary = fields.Properties().ToDictionary(p => p.Name, p => new LazyDocument(p.Value, s, r));
 				hit.Fields = new FieldValues(settings.Inferrer, fieldsDictionary);
 			}
 
diff --git a/src/Nest/XPack/Sql/QuerySql/SqlValue.cs b/src/Nest/XPack/Sql/QuerySql/SqlValue.cs
@@ -8,16 +8,18 @@ namespace Nest
 	[ContractJsonConverter(typeof(SqlValueJsonConverter))]
 	public class SqlValue : LazyDocument
 	{
-		internal SqlValue(JToken token, IElasticsearchSerializer serializer) : base(token, serializer) { }
+		internal SqlValue(JToken token, IElasticsearchSerializer sourceSerializer, IElasticsearchSerializer requestResponseSerializer)
+			: base(token, sourceSerializer, requestResponseSerializer) { }
 	}
 
 	internal class SqlValueJsonConverter : LazyDocumentJsonConverter
 	{
 		public override object ReadJson(JsonReader reader, Type objectType, object existingValue, JsonSerializer serializer)
 		{
 			var sourceSerializer = serializer.GetConnectionSettings().SourceSerializer;
+			var requestResponseSerializer = serializer.GetConnectionSettings().RequestResponseSerializer;
 			var token = reader.ReadTokenWithDateParseHandlingNone();
-			return new SqlValue(token, sourceSerializer);
+			return new SqlValue(token, sourceSerializer, requestResponseSerializer);
 		}
 	}
 }
diff --git a/src/Tests/Tests/Document/Multiple/Bulk/BulkSourceDeserialize.cs b/src/Tests/Tests/Document/Multiple/Bulk/BulkSourceDeserialize.cs
@@ -0,0 +1,71 @@
+using System;
+using System.Linq;
+using System.Text;
+using Elastic.Xunit.XunitPlumbing;
+using Elasticsearch.Net;
+using FluentAssertions;
+using Nest;
+using Tests.Core.Extensions;
+
+namespace Tests.Document.Multiple.Bulk
+{
+    public class BulkSourceDeserialize
+    {
+        [U]
+        public void CanDeserialize()
+        {
+            var json = @"{
+                            ""took"": 61,
+                            ""errors"": false,
+                            ""items"": [{
+                                ""update"": {
+                                    ""_index"": ""test"",
+                                    ""_type"": ""_doc"",
+                                    ""_id"": ""1"",
+                                    ""_version"": 2,
+                                    ""result"": ""updated"",
+                                    ""_shards"": {
+                                        ""total"": 2,
+                                        ""successful"": 1,
+                                        ""failed"": 0
+                                    },
+                                    ""_seq_no"": 3,
+                                    ""_primary_term"": 1,
+                                    ""get"": {
+                                        ""_seq_no"": 3,
+                                        ""_primary_term"": 1,
+                                        ""found"": true,
+                                        ""_source"": {
+                                            ""field1"": ""value1"",
+                                            ""field2"": ""value2""
+                                        }
+                                    },
+                                    ""status"": 200
+                                }
+                            }]
+                        }";
+
+            var pool = new SingleNodeConnectionPool(new Uri("http://localhost:9200"));
+
+            var connection = new InMemoryConnection(Encoding.UTF8.GetBytes(json));
+            var settings = new ConnectionSettings(pool, connection);
+            var client = new ElasticClient(settings);
+
+            var bulkResponse = client.Bulk(r => r);
+
+            bulkResponse.ShouldBeValid();
+
+            var simpleObject = bulkResponse.Items.First().GetResponse<SimpleObject>();
+
+            simpleObject.Found.Should().BeTrue();
+            simpleObject.Source.field1.Should().Be("value1");
+            simpleObject.Source.field2.Should().Be("value2");
+        }
+
+        private class SimpleObject
+        {
+            public string field1 { get; set; }
+            public string field2 { get; set; }
+        }
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -187,8 +187,9 @@ private IAggregate GetTopHitsAggregate(JsonReader reader, JsonSerializer seriali`
`187`	`187`	`var hits = o[Parser.Hits].Children().OfType<JObject>();`
`188`	`188`	`reader.Read();`
`189`	`189`	`//using request/response serializer here because doc is wrapped in NEST's Hit<T>`
`190`		`- var s = serializer.GetConnectionSettings().RequestResponseSerializer;`
`191`		`- var lazyHits = hits.Select(h => new LazyDocument(h, s)).ToList();`
	`190`	`+ var s = serializer.GetConnectionSettings().SourceSerializer;`
	`191`	`+ var r = serializer.GetConnectionSettings().RequestResponseSerializer;`
	`192`	`+ var lazyHits = hits.Select(h => new LazyDocument(h, s, r)).ToList();`
`192`	`193`	`return new TopHitsAggregate(lazyHits)`
`193`	`194`	`{`
`194`	`195`	`Total = total,`
`@@ -573,7 +574,8 @@ private IAggregate GetValueAggregate(JsonReader reader, JsonSerializer serialize`
`573`	`574`	`if (scriptedMetric != null)`
`574`	`575`	`{`
`575`	`576`	`var s = serializer.GetConnectionSettings().SourceSerializer;`
`576`		`- return new ScriptedMetricAggregate(new LazyDocument(scriptedMetric, s));`
	`577`	`+ var r = serializer.GetConnectionSettings().RequestResponseSerializer;`
	`578`	`+ return new ScriptedMetricAggregate(new LazyDocument(scriptedMetric, s, r));`
`577`	`579`	`}`
`578`	`580`	`return valueMetric;`
`579`	`581`	`}`
Original file line number	Diff line number	Diff line change
`@@ -28,12 +28,14 @@ public interface ILazyDocument`
`28`	`28`	`/// <inheritdoc />`
`29`	`29`	`public class LazyDocument : ILazyDocument`
`30`	`30`	`{`
`31`		`- private readonly IElasticsearchSerializer _serializer;`
	`31`	`+ private readonly IElasticsearchSerializer _sourceSerializer;`
	`32`	`+ private readonly IElasticsearchSerializer _requestResponseSerializer;`
`32`	`33`
`33`		`- internal LazyDocument(JToken token, IElasticsearchSerializer serializer)`
	`34`	`+ internal LazyDocument(JToken token, IElasticsearchSerializer sourceSerializer, IElasticsearchSerializer requestResponseSerializer)`
`34`	`35`	`{`
`35`	`36`	`Token = token;`
`36`		`- _serializer = serializer;`
	`37`	`+ _sourceSerializer = sourceSerializer;`
	`38`	`+ _requestResponseSerializer = requestResponseSerializer;`
`37`	`39`	`}`
`38`	`40`
`39`	`41`	`internal JToken Token { get; }`
`@@ -43,15 +45,22 @@ public T As<T>()`
`43`	`45`	`{`
`44`	`46`	`if (Token == null) return default(T);`
`45`	`47`	`using (var ms = Token.ToStream())`
`46`		`- return _serializer.Deserialize<T>(ms);`
	`48`	`+ return _sourceSerializer.Deserialize<T>(ms);`
`47`	`49`	`}`
`48`	`50`
`49`	`51`	`/// <inheritdoc />`
`50`	`52`	`public object As(Type objectType)`
`51`	`53`	`{`
`52`	`54`	`if (Token == null) return null;`
`53`	`55`	`using (var ms = Token.ToStream())`
`54`		`- return _serializer.Deserialize(objectType, ms);`
	`56`	`+ return _sourceSerializer.Deserialize(objectType, ms);`
	`57`	`+ }`
	`58`	`+`
	`59`	`+ internal T AsUsingRequestResponseSerializer<T>()`
	`60`	`+ {`
	`61`	`+ if (Token == null) return default(T);`
	`62`	`+ using (var ms = Token.ToStream())`
	`63`	`+ return _requestResponseSerializer.Deserialize<T>(ms);`
`55`	`64`	`}`
`56`	`65`	`}`
`57`	`66`	`}`
Original file line number	Diff line number	Diff line change
`@@ -20,8 +20,9 @@ public override void WriteJson(JsonWriter writer, object value, JsonSerializer s`
`20`	`20`	`public override object ReadJson(JsonReader reader, Type objectType, object existingValue, JsonSerializer serializer)`
`21`	`21`	`{`
`22`	`22`	`var sourceSerializer = serializer.GetConnectionSettings().SourceSerializer;`
	`23`	`+ var requestResponseSerializer = serializer.GetConnectionSettings().RequestResponseSerializer;`
`23`	`24`	`var token = reader.ReadTokenWithDateParseHandlingNone();`
`24`		`- return new LazyDocument(token, sourceSerializer);`
	`25`	`+ return new LazyDocument(token, sourceSerializer, requestResponseSerializer);`
`25`	`26`	`}`
`26`	`27`
`27`	`28`	`public override bool CanConvert(Type objectType) => true;`
Original file line number	Diff line number	Diff line change
`@@ -28,5 +28,8 @@ internal class BulkUpdateBody<TDocument, TPartialUpdate>`
`28`	`28`
`29`	`29`	`[JsonProperty("if_primary_term")]`
`30`	`30`	`internal long? IfPrimaryTerm { get; set; }`
	`31`	`+`
	`32`	`+ [JsonProperty("_source")]`
	`33`	`+ internal Union<bool, ISourceFilter> Source { get; set; }`
`31`	`34`	`}`
`32`	`35`	`}`
Original file line number	Diff line number	Diff line change
`@@ -27,10 +27,11 @@ private static void CreateMultiHit<T>(MultiHitTuple tuple, JsonSerializer serial`
`27`	`27`	`var hit = tuple.Hit.ToObject<MultiGetHit<T>>(serializer);`
`28`	`28`	`var settings = serializer.GetConnectionSettings();`
`29`	`29`	`var s = serializer.GetConnectionSettings().SourceSerializer;`
	`30`	`+ var r = serializer.GetConnectionSettings().RequestResponseSerializer;`
`30`	`31`
`31`	`32`	`if (tuple.Hit["fields"] is JObject fields)`
`32`	`33`	`{`
`33`		`- var fieldsDictionary = fields.Properties().ToDictionary(p => p.Name, p => new LazyDocument(p.Value, s));`
	`34`	`+ var fieldsDictionary = fields.Properties().ToDictionary(p => p.Name, p => new LazyDocument(p.Value, s, r));`
`34`	`35`	`hit.Fields = new FieldValues(settings.Inferrer, fieldsDictionary);`
`35`	`36`	`}`
`36`	`37`
Original file line number	Diff line number	Diff line change
`@@ -8,16 +8,18 @@ namespace Nest`
`8`	`8`	`[ContractJsonConverter(typeof(SqlValueJsonConverter))]`
`9`	`9`	`public class SqlValue : LazyDocument`
`10`	`10`	`{`
`11`		`- internal SqlValue(JToken token, IElasticsearchSerializer serializer) : base(token, serializer) { }`
	`11`	`+ internal SqlValue(JToken token, IElasticsearchSerializer sourceSerializer, IElasticsearchSerializer requestResponseSerializer)`
	`12`	`+ : base(token, sourceSerializer, requestResponseSerializer) { }`
`12`	`13`	`}`
`13`	`14`
`14`	`15`	`internal class SqlValueJsonConverter : LazyDocumentJsonConverter`
`15`	`16`	`{`
`16`	`17`	`public override object ReadJson(JsonReader reader, Type objectType, object existingValue, JsonSerializer serializer)`
`17`	`18`	`{`
`18`	`19`	`var sourceSerializer = serializer.GetConnectionSettings().SourceSerializer;`
	`20`	`+ var requestResponseSerializer = serializer.GetConnectionSettings().RequestResponseSerializer;`
`19`	`21`	`var token = reader.ReadTokenWithDateParseHandlingNone();`
`20`		`- return new SqlValue(token, sourceSerializer);`
	`22`	`+ return new SqlValue(token, sourceSerializer, requestResponseSerializer);`
`21`	`23`	`}`
`22`	`24`	`}`
`23`	`25`	`}`