Add optimistic concurrency control

This commit adds optimistic concurrency control as described in https://www.elastic.co/guide/en/elasticsearch/reference/7.3/optimistic-concurrency-control.html. Close olivere#1143 Close olivere#1154
neoreids · Aug 5, 2019 · 6da3cc8 · 6da3cc8
1 parent 37f9df1
commit 6da3cc8
Show file tree

Hide file tree

Showing 20 changed files with 611 additions and 174 deletions.
diff --git a/bulk_delete_request.go b/bulk_delete_request.go
@@ -20,13 +20,15 @@ import (
 // for details.
 type BulkDeleteRequest struct {
 	BulkableRequest
-	index       string
-	typ         string
-	id          string
-	parent      string
-	routing     string
-	version     int64  // default is MATCH_ANY
-	versionType string // default is "internal"
+	index         string
+	typ           string
+	id            string
+	parent        string
+	routing       string
+	version       int64  // default is MATCH_ANY
+	versionType   string // default is "internal"
+	ifSeqNo       *int64
+	ifPrimaryTerm *int64
 
 	source []string
 
@@ -38,13 +40,15 @@ type bulkDeleteRequestCommand map[string]bulkDeleteRequestCommandOp
 
 //easyjson:json
 type bulkDeleteRequestCommandOp struct {
-	Index       string `json:"_index,omitempty"`
-	Type        string `json:"_type,omitempty"`
-	Id          string `json:"_id,omitempty"`
-	Parent      string `json:"parent,omitempty"`
-	Routing     string `json:"routing,omitempty"`
-	Version     int64  `json:"version,omitempty"`
-	VersionType string `json:"version_type,omitempty"`
+	Index         string `json:"_index,omitempty"`
+	Type          string `json:"_type,omitempty"`
+	Id            string `json:"_id,omitempty"`
+	Parent        string `json:"parent,omitempty"`
+	Routing       string `json:"routing,omitempty"`
+	Version       int64  `json:"version,omitempty"`
+	VersionType   string `json:"version_type,omitempty"`
+	IfSeqNo       *int64 `json:"if_seq_no,omitempty"`
+	IfPrimaryTerm *int64 `json:"if_primary_term,omitempty"`
 }
 
 // NewBulkDeleteRequest returns a new BulkDeleteRequest.
@@ -116,6 +120,20 @@ func (r *BulkDeleteRequest) VersionType(versionType string) *BulkDeleteRequest {
 	return r
 }
 
+// IfSeqNo indicates to only perform the delete operation if the last
+// operation that has changed the document has the specified sequence number.
+func (r *BulkDeleteRequest) IfSeqNo(ifSeqNo int64) *BulkDeleteRequest {
+	r.ifSeqNo = &ifSeqNo
+	return r
+}
+
+// IfPrimaryTerm indicates to only perform the delete operation if the
+// last operation that has changed the document has the specified primary term.
+func (r *BulkDeleteRequest) IfPrimaryTerm(ifPrimaryTerm int64) *BulkDeleteRequest {
+	r.ifPrimaryTerm = &ifPrimaryTerm
+	return r
+}
+
 // String returns the on-wire representation of the delete request,
 // concatenated as a single string.
 func (r *BulkDeleteRequest) String() string {
@@ -136,13 +154,15 @@ func (r *BulkDeleteRequest) Source() ([]string, error) {
 	}
 	command := bulkDeleteRequestCommand{
 		"delete": bulkDeleteRequestCommandOp{
-			Index:       r.index,
-			Type:        r.typ,
-			Id:          r.id,
-			Routing:     r.routing,
-			Parent:      r.parent,
-			Version:     r.version,
-			VersionType: r.versionType,
+			Index:         r.index,
+			Type:          r.typ,
+			Id:            r.id,
+			Routing:       r.routing,
+			Parent:        r.parent,
+			Version:       r.version,
+			VersionType:   r.versionType,
+			IfSeqNo:       r.ifSeqNo,
+			IfPrimaryTerm: r.ifPrimaryTerm,
 		},
 	}
 

diff --git a/bulk_delete_request_easyjson.go b/bulk_delete_request_easyjson.go
diff --git a/bulk_index_request.go b/bulk_index_request.go
@@ -29,6 +29,8 @@ type BulkIndexRequest struct {
 	doc             interface{}
 	pipeline        string
 	retryOnConflict *int
+	ifSeqNo         *int64
+	ifPrimaryTerm   *int64
 
 	source []string
 
@@ -50,6 +52,8 @@ type bulkIndexRequestCommandOp struct {
 	Version         *int64 `json:"version,omitempty"`
 	VersionType     string `json:"version_type,omitempty"`
 	Pipeline        string `json:"pipeline,omitempty"`
+	IfSeqNo         *int64 `json:"if_seq_no,omitempty"`
+	IfPrimaryTerm   *int64 `json:"if_primary_term,omitempty"`
 }
 
 // NewBulkIndexRequest returns a new BulkIndexRequest.
@@ -158,6 +162,20 @@ func (r *BulkIndexRequest) Pipeline(pipeline string) *BulkIndexRequest {
 	return r
 }
 
+// IfSeqNo indicates to only perform the index operation if the last
+// operation that has changed the document has the specified sequence number.
+func (r *BulkIndexRequest) IfSeqNo(ifSeqNo int64) *BulkIndexRequest {
+	r.ifSeqNo = &ifSeqNo
+	return r
+}
+
+// IfPrimaryTerm indicates to only perform the index operation if the
+// last operation that has changed the document has the specified primary term.
+func (r *BulkIndexRequest) IfPrimaryTerm(ifPrimaryTerm int64) *BulkIndexRequest {
+	r.ifPrimaryTerm = &ifPrimaryTerm
+	return r
+}
+
 // String returns the on-wire representation of the index request,
 // concatenated as a single string.
 func (r *BulkIndexRequest) String() string {
@@ -193,6 +211,8 @@ func (r *BulkIndexRequest) Source() ([]string, error) {
 		VersionType:     r.versionType,
 		RetryOnConflict: r.retryOnConflict,
 		Pipeline:        r.pipeline,
+		IfSeqNo:         r.ifSeqNo,
+		IfPrimaryTerm:   r.ifPrimaryTerm,
 	}
 	command := bulkIndexRequestCommand{
 		r.opType: indexCommand,