File Coverage

blib/lib/MongoDB/BulkWrite.pm

Criterion	Covered	Total	%
statement	33	76	43.4
branch	0	22	0.0
condition	0	15	0.0
subroutine	11	20	55.0
pod	3	3	100.0
total	47	136	34.5

#  Copyright 2014 - present MongoDB, Inc.
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#  http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.

use strict;
use warnings;
package MongoDB::BulkWrite;

# ABSTRACT: MongoDB bulk write interface

use version;
our $VERSION = 'v2.2.0';

use MongoDB::Error;
use MongoDB::Op::_BulkWrite;
use MongoDB::BulkWriteResult;
use MongoDB::BulkWriteView;

use Moo;
use MongoDB::_Types qw(
    Boolish
    to_WriteConcern
);
use Types::Standard qw(
    ArrayRef
    InstanceOf
);
use namespace::clean -except => 'meta';

#pod =attr collection (required)
#pod
#pod The L where the operations are to be performed.
#pod
#pod =cut

has 'collection' => (
    is       => 'ro',
    isa      => InstanceOf['MongoDB::Collection'],
    required => 1,
);

#pod =attr ordered (required)
#pod
#pod A boolean for whether or not operations should be ordered (true) or
#pod unordered (false).
#pod
#pod =cut

has 'ordered' => (
    is       => 'ro',
    isa      => Boolish,
    required => 1,
);

#pod =attr bypassDocumentValidation
#pod
#pod A boolean for whether or not operations should bypass document validation.
#pod Default is false.
#pod
#pod =cut

has 'bypassDocumentValidation' => (
    is       => 'ro',
    isa      => Boolish,
);

has '_executed' => (
    is       => 'rw',
    isa      => Boolish,
    init_arg => undef,
    default  => 0,
);

has '_queue' => (
    is       => 'rw',
    isa      => ArrayRef[ArrayRef],
    init_arg => undef,
    default  => sub { [] },
);

sub _enqueue_write {
    my $self = shift;
    push @{$self->{_queue}}, @_;
}

sub _all_writes { return @{$_[0]->{_queue}} }

sub _count_writes { return scalar @{$_[0]->{_queue}} }

sub _clear_writes { @{$_[0]->{_queue}} = (); return; }

has '_database' => (
    is         => 'lazy',
    isa        => InstanceOf['MongoDB::Database'],
);

sub _build__database {
    my ($self) = @_;
    return $self->collection->database;
}

has '_client' => (
    is         => 'lazy',
    isa        => InstanceOf['MongoDB::MongoClient'],
);

sub _build__client {
    my ($self) = @_;
    return $self->_database->_client;
}

has _retryable => (
    is => 'rw',
    isa => Boolish,
    default => 1,
);

#pod =method find
#pod
#pod     $view = $bulk->find( $query_document );
#pod
#pod The C method returns a L object that allows
#pod write operations like C or C, constrained by a query document.
#pod
#pod A query document is required.  Use an empty hashref for no criteria:
#pod
#pod     $bulk->find( {} )->remove; # remove all documents!
#pod
#pod An exception will be thrown on error.
#pod
#pod =cut

sub find {
    my ( $self, $doc ) = @_;

    MongoDB::UsageError->throw("find requires a criteria document. Use an empty hashref for no criteria.")
      unless defined $doc;

    my $type = ref $doc;
    unless ( @_ == 2 && grep { $type eq $_ } qw/HASH ARRAY Tie::IxHash/ ) {
        MongoDB::UsageError->throw("argument to find must be a single hashref, arrayref or Tie::IxHash");
    }

    if ( ref $doc eq 'ARRAY' ) {
        MongoDB::UsageError->throw("array reference to find must have key/value pairs")
          if @$doc % 2;
        $doc = {@$doc};
    }

    return MongoDB::BulkWriteView->new(
        _query => $doc,
        _bulk  => $self,
    );
}

#pod =method insert_one
#pod
#pod     $bulk->insert_one( $doc );
#pod
#pod Queues a document for insertion when L is called.  The document may
#pod be a hash reference, an array reference (with balanced key/value pairs) or a
#pod L object.  If the document does not have an C<_id> field, one will
#pod be added to the original.
#pod
#pod The method has an empty return on success; an exception will be thrown on error.
#pod
#pod =cut

sub insert_one {
    MongoDB::UsageError->throw("insert_one requires a single document reference as an argument")
      unless @_ == 2 && ref( $_[1] );

    my ( $self, $doc ) = @_;

    if ( ref $doc eq 'ARRAY' ) {
        MongoDB::UsageError->throw("array reference to find must have key/value pairs")
          if @$doc % 2;
        $doc = {@$doc};
    }

    $self->_enqueue_write( [ insert => $doc ] );

    return;
}

#pod =method execute
#pod
#pod     my $result = $bulk->execute;
#pod     # Optional write concern:
#pod     my $result = $bulk->execute( $concern );
#pod     # With options
#pod     my $result = $bulk->execute( $concern, $options );
#pod
#pod Executes the queued operations.  The order and semantics depend on
#pod whether the bulk object is ordered or unordered:
#pod
#pod =for :list
#pod * ordered — operations are executed in order, but operations of the same type
#pod   (e.g. multiple inserts) may be grouped together and sent to the server.  If
#pod   the server returns an error, the bulk operation will stop and an error will
#pod   be thrown.
#pod * unordered — operations are grouped by type and sent to the server in an
#pod   unpredictable order.  After all operations are sent, if any errors occurred,
#pod   an error will be thrown.
#pod
#pod When grouping operations of a type, operations will be sent to the server in
#pod batches not exceeding 16MiB or 1000 items (for a version 2.6 or later server)
#pod or individually (for legacy servers without write command support).
#pod
#pod A write concern is optional, and can either take a pre-constructed WriteConcern
#pod object, or the arguments to construct one.  For information on write concerns,
#pod see L.
#pod
#pod The options argument is an optional hashref which can contain the following
#pod values:
#pod
#pod =for :list
#pod * C - the session to use for these operations. If not supplied, will
#pod   use an implicit session. For more information see L
#pod
#pod This method returns a L object if the bulk operation
#pod executes successfully.
#pod
#pod Typical errors might include:
#pod
#pod =for :list
#pod * C — one or more write operations failed
#pod * C - all writes were accepted by a primary, but
#pod   the write concern failed
#pod * C — a command to the database failed entirely
#pod
#pod See L for more on error handling.
#pod
#pod B: it is an error to call C without any operations or
#pod to call C more than once on the same bulk object.
#pod
#pod
#pod =cut

sub execute {
    my ( $self, $write_concern, $options ) = @_;
    $write_concern = to_WriteConcern($write_concern)
        if defined($write_concern) && ref($write_concern) ne 'MongoDB::WriteConcern';

    if ( $self->_executed ) {
        MongoDB::UsageError->throw("bulk op execute called more than once");
    }
    else {
        $self->_executed(1);
    }

    unless ( $self->_count_writes ) {
        MongoDB::UsageError->throw("no bulk ops to execute");
    }

    $write_concern ||= $self->collection->write_concern;

    my $session = $self->_client->_get_session_from_hashref( $options );

    # Transaction write concern overrides all
    $write_concern = $session->_get_transaction_write_concern
      if defined $session && $session->_active_transaction;

    my $op = MongoDB::Op::_BulkWrite->_new(
        client                   => $self->_client,
        db_name                  => $self->_database->name,
        coll_name                => $self->collection->name,
        full_name                => $self->collection->full_name,
        queue                    => $self->_queue,
        ordered                  => $self->ordered,
        bypassDocumentValidation => $self->bypassDocumentValidation,
        bson_codec               => $self->collection->bson_codec,
        write_concern            => $write_concern,
        session                  => $session,
        monitoring_callback      => $self->_client->monitoring_callback,
        _retryable               => $self->_retryable,
    );

    # Op::_BulkWrite internally does retryable writes
    return $self->_client->send_write_op( $op );
}

1;

__END__