Add if_seq_no and if_primary_term to bulk updates (#4015)

(cherry picked from commit d822c33)

Author: codebrain

src/Nest/Document/Multiple/Bulk/BulkOperation/BulkIndex.cs

@@ -13,6 +13,12 @@ public interface IBulkIndexOperation<T> : IBulkOperation
 
 		[JsonProperty("pipeline")]
 		string Pipeline { get; set; }
+		
+		[JsonProperty("if_seq_no")]
+		long? IfSequenceNumber { get; set; }
+
+		[JsonProperty("if_primary_term")]
+		long? IfPrimaryTerm { get; set; }
 	}
 
 	public class BulkIndexOperation<T> : BulkOperationBase, IBulkIndexOperation<T>
@@ -25,6 +31,10 @@ public class BulkIndexOperation<T> : BulkOperationBase, IBulkIndexOperation<T>
 		public string Percolate { get; set; }
 
 		public string Pipeline { get; set; }
+		
+		public long? IfSequenceNumber { get; set; }
+
+		public long? IfPrimaryTerm { get; set; }
 
 		protected override Type ClrType => typeof(T);
 
@@ -46,6 +56,8 @@ public class BulkIndexDescriptor<T> : BulkOperationDescriptorBase<BulkIndexDescr
 		T IBulkIndexOperation<T>.Document { get; set; }
 		string IBulkIndexOperation<T>.Percolate { get; set; }
 		string IBulkIndexOperation<T>.Pipeline { get; set; }
+		long? IBulkIndexOperation<T>.IfSequenceNumber { get; set; }
+		long? IBulkIndexOperation<T>.IfPrimaryTerm { get; set; }
 
 		protected override object GetBulkOperationBody() => Self.Document;
 
@@ -64,5 +76,9 @@ public class BulkIndexDescriptor<T> : BulkOperationDescriptorBase<BulkIndexDescr
 		public BulkIndexDescriptor<T> Pipeline(string pipeline) => Assign(pipeline, (a, v) => a.Pipeline = v);
 
 		public BulkIndexDescriptor<T> Percolate(string percolate) => Assign(percolate, (a, v) => a.Percolate = v);
+		
+		public BulkIndexDescriptor<T> IfSequenceNumber(long? seqNo) => Assign(seqNo, (a, v) => a.IfSequenceNumber = v);
+
+		public BulkIndexDescriptor<T> IfPrimaryTerm(long? primaryTerm) => Assign(primaryTerm, (a, v) => a.IfSequenceNumber = v);
 	}
 }

src/Nest/Document/Multiple/Bulk/BulkOperation/BulkOperationBase.cs

@@ -62,7 +62,6 @@ public abstract class BulkOperationDescriptorBase<TDescriptor, TInterface>
 		/// <summary>
 		/// Only used for bulk update operations but in the future might come in handy for other complex bulk ops.
 		/// </summary>
-		/// <returns></returns>
 		object IBulkOperation.GetBody() => GetBulkOperationBody();
 
 		Id IBulkOperation.GetIdForOperation(Inferrer inferrer) => GetIdForOperation(inferrer);

src/Nest/Document/Multiple/Bulk/BulkOperation/BulkUpdate.cs

@@ -40,6 +40,10 @@ public interface IBulkUpdateOperation<TDocument, TPartialDocument> : IBulkOperat
 		/// A document to upsert when the specified document to be updated is not found
 		/// </summary>
 		TDocument Upsert { get; set; }
+
+		long? IfSequenceNumber { get; set; }
+
+		long? IfPrimaryTerm { get; set; }
 	}
 
 	public class BulkUpdateOperation<TDocument, TPartialDocument> : BulkOperationBase, IBulkUpdateOperation<TDocument, TPartialDocument>
@@ -108,6 +112,10 @@ public BulkUpdateOperation(TDocument idFrom, TPartialDocument update, bool useId
 		/// </summary>
 		public TDocument Upsert { get; set; }
 
+		public long? IfSequenceNumber { get; set; }
+
+		public long? IfPrimaryTerm { get; set; }
+
 		protected override Type ClrType => typeof(TDocument);
 
 		protected override string Operation => "update";
@@ -125,7 +133,9 @@ protected override object GetBody() =>
 				_Script = Script,
 				_Upsert = Upsert,
 				_DocAsUpsert = DocAsUpsert,
-				_ScriptedUpsert = ScriptedUpsert
+				_ScriptedUpsert = ScriptedUpsert,
+				IfPrimaryTerm = IfPrimaryTerm,
+				IfSequenceNumber = IfSequenceNumber
 			};
 	}
 
@@ -139,11 +149,12 @@ public class BulkUpdateDescriptor<TDocument, TPartialDocument>
 		protected override string BulkOperationType => "update";
 		TPartialDocument IBulkUpdateOperation<TDocument, TPartialDocument>.Doc { get; set; }
 		bool? IBulkUpdateOperation<TDocument, TPartialDocument>.DocAsUpsert { get; set; }
-
 		TDocument IBulkUpdateOperation<TDocument, TPartialDocument>.IdFrom { get; set; }
 		IScript IBulkUpdateOperation<TDocument, TPartialDocument>.Script { get; set; }
 		bool? IBulkUpdateOperation<TDocument, TPartialDocument>.ScriptedUpsert { get; set; }
 		TDocument IBulkUpdateOperation<TDocument, TPartialDocument>.Upsert { get; set; }
+		long? IBulkUpdateOperation<TDocument, TPartialDocument>.IfSequenceNumber { get; set; }
+		long? IBulkUpdateOperation<TDocument, TPartialDocument>.IfPrimaryTerm { get; set; }
 
 		protected override object GetBulkOperationBody() =>
 			new BulkUpdateBody<TDocument, TPartialDocument>
@@ -152,7 +163,9 @@ protected override object GetBulkOperationBody() =>
 				_Script = Self.Script,
 				_Upsert = Self.Upsert,
 				_DocAsUpsert = Self.DocAsUpsert,
-				_ScriptedUpsert = Self.ScriptedUpsert
+				_ScriptedUpsert = Self.ScriptedUpsert,
+				IfPrimaryTerm = Self.IfPrimaryTerm,
+				IfSequenceNumber = Self.IfSequenceNumber
 			};
 
 		protected override Id GetIdForOperation(Inferrer inferrer) =>
@@ -209,5 +222,17 @@ public BulkUpdateDescriptor<TDocument, TPartialDocument> Script(Func<ScriptDescr
 		/// </summary>
 		public BulkUpdateDescriptor<TDocument, TPartialDocument> RetriesOnConflict(int? retriesOnConflict) =>
 			Assign(retriesOnConflict, (a, v) => a.RetriesOnConflict = v);
+
+		/// <summary>
+		/// Operations can be made conditional and only be performed if the last modification to the document was assigned the sequence number.
+		/// </summary>
+		public BulkUpdateDescriptor<TDocument, TPartialDocument> IfSequenceNumber(long? seqNo) =>
+			Assign(seqNo, (a, v) => a.IfSequenceNumber = v);
+
+		/// <summary>
+		/// Operations can be made conditional and only be performed if the last modification to the document was assigned the primary term.
+		/// </summary>
+		public BulkUpdateDescriptor<TDocument, TPartialDocument> IfPrimaryTerm(long? primaryTerm) =>
+			Assign(primaryTerm, (a, v) => a.IfSequenceNumber = v);
 	}
 }

src/Nest/Document/Multiple/Bulk/BulkOperation/BulkUpdateBody.cs

@@ -19,8 +19,14 @@ internal class BulkUpdateBody<TDocument, TPartialUpdate>
 		[JsonProperty("scripted_upsert")]
 		internal bool? _ScriptedUpsert { get; set; }
 
+		[JsonProperty("if_seq_no")]
+		internal long? IfSequenceNumber { get; set; }
+
 		[JsonProperty("upsert")]
 		[JsonConverter(typeof(CollapsedSourceConverter))]
 		internal TDocument _Upsert { get; set; }
+
+		[JsonProperty("if_primary_term")]
+		internal long? IfPrimaryTerm { get; set; }
 	}
 }