bulk

Documentation

Overview

Package bulk contains methods and structures for handling Elasticsearch bulk operations.

Index

• Variables
• func NewRequest(ctx context.Context, client *elasticsearch.TypedClient, options ...Option) (chan Operation, chan Response)
• type OpType
• type Operation
  • func NewOperation(doc any, options ...OperationOption) Operation
• type OperationOption
  • func AsOperationType(opType OpType) OperationOption
  • func SetDocID(id string) OperationOption
  • func ToIndex(index string) OperationOption
  • func Upsert(value bool) OperationOption
• type OperationResponse
  • func GetOperationResponses(resp []map[operationtype.OperationType]types.ResponseItem) []*OperationResponse
  • func (r *OperationResponse) Created() bool
  • func (r *OperationResponse) Deleted() bool
  • func (r *OperationResponse) State() (string, error)
  • func (r *OperationResponse) Updated() bool
• type Option
  • func WithIndex(index string) Option
  • func WithPipeline(pipeline string) Option
• type Request
  • func (r *Request) AddOperation(operation Operation) error
• type Response
  • func (r *Response) FailedDocs() []string

Variables

var (
	// ErrCreateOpFailed indicates creating a bulk operation failed.
	ErrCreateOpFailed = errors.New("could not create bulk operation")
	// ErrOpFailed indicates executing a bulk operation failed.
	ErrOpFailed = errors.New("bulk operation failed")
	// ErrBulkHasErrors indicates the bulk request completed but some operations produced an error.
	ErrBulkHasErrors = errors.New("bulk request completed with some operation errors")
)

Functions

func NewRequest

func NewRequest(
	ctx context.Context,
	client *elasticsearch.TypedClient,
	options ...Option,
) (chan Operation, chan Response)

NewRequest creates a new bulk requesst object with the given options. After creation, document operations can be added with the AddOperations method.

Types

type OpType

type OpType int

OpType represents the type of bulk operation to perform.

const (
	// BulkCreate is a create operation.
	BulkCreate OpType = iota
	// BulkIndex is a index operation.
	BulkIndex
	// BulkDelete is a delete operation.
	BulkDelete
	// BulkUpdate is an update operation.
	BulkUpdate
)

type Operation

type Operation struct {
	// contains filtered or unexported fields
}

Operation represents an individual document's bulk operation.

func NewOperation

func NewOperation(doc any, options ...OperationOption) Operation

NewOperation creates a new bulk operation for a document with the given options.

type OperationOption

type OperationOption func(*Operation)

OperationOption is a functional option for a bulk operation.

func AsOperationType

func AsOperationType(opType OpType) OperationOption

AsOperationType option specifies the type of bulk operation to perform. If this option is not specified, the operation will default to a create operation.

func SetDocID

func SetDocID(id string) OperationOption

SetDocID option sets the doc id for the operation.

func ToIndex

func ToIndex(index string) OperationOption

ToIndex option sets the index containing the document.

func Upsert

func Upsert(value bool) OperationOption

Upsert option will perform the bulk action as an upsert. Only used where operation type is update.

type OperationResponse

type OperationResponse struct {
	*types.ResponseItem
}

OperationResponse holds details about a bulk operation's result.

func GetOperationResponses

func GetOperationResponses(resp []map[operationtype.OperationType]types.ResponseItem) []*OperationResponse

GetOperationResponses extracts a slice of OperationResponse from the bulk request response data.

func (*OperationResponse) Created

func (r *OperationResponse) Created() bool

Created indicates the document was created.

func (*OperationResponse) Deleted

func (r *OperationResponse) Deleted() bool

Deleted indicates the document was deleted.

func (*OperationResponse) State

func (r *OperationResponse) State() (string, error)

State returns a string indicating the operation result and a non-nil error if one occurred.

func (*OperationResponse) Updated

func (r *OperationResponse) Updated() bool

Updated indicates the document was updated.

type Option

type Option func(*Request)

Option is a functional option for a bulk request.

func WithIndex

func WithIndex(index string) Option

WithIndex defines the index on which the request will operate.

func WithPipeline

func WithPipeline(pipeline string) Option

WithPipeline defines the ingest pipeline to use on each document.

type Request

type Request struct {
	*bulk.Bulk
}

Request represents a bulk request.

func (*Request) AddOperation

func (r *Request) AddOperation(operation Operation) error

AddOperation adds document operations to the bulk request.

type Response

type Response struct {
	Err       error
	Responses []*OperationResponse
}

Response holds details about a bulk request's results.

func (*Response) FailedDocs

func (r *Response) FailedDocs() []string

FailedDocs returns the doc IDs of all documents whose operation failed.