[Summary view]

Differences Between: [Versions 311 and 401]

   1  <?php
   2  /*
   3   * Copyright 2015-2017 MongoDB, Inc.
   4   *
   5   * Licensed under the Apache License, Version 2.0 (the "License");
   6   * you may not use this file except in compliance with the License.
   7   * You may obtain a copy of the License at
   8   *
   9   *   http://www.apache.org/licenses/LICENSE-2.0
  10   *
  11   * Unless required by applicable law or agreed to in writing, software
  12   * distributed under the License is distributed on an "AS IS" BASIS,
  13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14   * See the License for the specific language governing permissions and
  15   * limitations under the License.
  16   */
  17  
  18  namespace MongoDB\Operation;
  19  
  20  use MongoDB\BulkWriteResult;
  21  use MongoDB\Driver\BulkWrite as Bulk;
  22  use MongoDB\Driver\Exception\RuntimeException as DriverRuntimeException;
  23  use MongoDB\Driver\Server;
  24  use MongoDB\Driver\Session;
  25  use MongoDB\Driver\WriteConcern;
  26  use MongoDB\Exception\InvalidArgumentException;
  27  use MongoDB\Exception\UnsupportedException;
  28  use function array_key_exists;
  29  use function count;
  30  use function current;
  31  use function is_array;
  32  use function is_bool;
  33  use function is_object;
  34  use function key;
  35  use function MongoDB\is_first_key_operator;
  36  use function MongoDB\is_pipeline;
  37  use function MongoDB\server_supports_feature;
  38  use function sprintf;
  39  
  40  /**
  41   * Operation for executing multiple write operations.
  42   *
  43   * @api
  44   * @see \MongoDB\Collection::bulkWrite()
  45   */
  46  class BulkWrite implements Executable
  47  {
  48      const DELETE_MANY = 'deleteMany';
  49      const DELETE_ONE  = 'deleteOne';
  50      const INSERT_ONE  = 'insertOne';
  51      const REPLACE_ONE = 'replaceOne';
  52      const UPDATE_MANY = 'updateMany';
  53      const UPDATE_ONE  = 'updateOne';
  54  
  55      /** @var integer */
  56      private static $wireVersionForArrayFilters = 6;
  57  
  58      /** @var integer */
  59      private static $wireVersionForCollation = 5;
  60  
  61      /** @var integer */
  62      private static $wireVersionForDocumentLevelValidation = 4;
  63  
  64      /** @var string */
  65      private $databaseName;
  66  
  67      /** @var string */
  68      private $collectionName;
  69  
  70      /** @var array[] */
  71      private $operations;
  72  
  73      /** @var array */
  74      private $options;
  75  
  76      /** @var boolean */
  77      private $isArrayFiltersUsed = false;
  78  
  79      /** @var boolean */
  80      private $isCollationUsed = false;
  81  
  82      /**
  83       * Constructs a bulk write operation.
  84       *
  85       * Example array structure for all supported operation types:
  86       *
  87       *  [
  88       *    [ 'deleteMany' => [ $filter, $options ] ],
  89       *    [ 'deleteOne'  => [ $filter, $options ] ],
  90       *    [ 'insertOne'  => [ $document ] ],
  91       *    [ 'replaceOne' => [ $filter, $replacement, $options ] ],
  92       *    [ 'updateMany' => [ $filter, $update, $options ] ],
  93       *    [ 'updateOne'  => [ $filter, $update, $options ] ],
  94       *  ]
  95       *
  96       * Arguments correspond to the respective Operation classes; however, the
  97       * writeConcern option is specified for the top-level bulk write operation
  98       * instead of each individual operation.
  99       *
 100       * Supported options for deleteMany and deleteOne operations:
 101       *
 102       *  * collation (document): Collation specification.
 103       *
 104       *    This is not supported for server versions < 3.4 and will result in an
 105       *    exception at execution time if used.
 106       *
 107       * Supported options for replaceOne, updateMany, and updateOne operations:
 108       *
 109       *  * collation (document): Collation specification.
 110       *
 111       *    This is not supported for server versions < 3.4 and will result in an
 112       *    exception at execution time if used.
 113       *
 114       *  * upsert (boolean): When true, a new document is created if no document
 115       *    matches the query. The default is false.
 116       *
 117       * Supported options for updateMany and updateOne operations:
 118       *
 119       *  * arrayFilters (document array): A set of filters specifying to which
 120       *    array elements an update should apply.
 121       *
 122       *    This is not supported for server versions < 3.6 and will result in an
 123       *    exception at execution time if used.
 124       *
 125       * Supported options for the bulk write operation:
 126       *
 127       *  * bypassDocumentValidation (boolean): If true, allows the write to
 128       *    circumvent document level validation. The default is false.
 129       *
 130       *    For servers < 3.2, this option is ignored as document level validation
 131       *    is not available.
 132       *
 133       *  * ordered (boolean): If true, when an insert fails, return without
 134       *    performing the remaining writes. If false, when a write fails,
 135       *    continue with the remaining writes, if any. The default is true.
 136       *
 137       *  * session (MongoDB\Driver\Session): Client session.
 138       *
 139       *    Sessions are not supported for server versions < 3.6.
 140       *
 141       *  * writeConcern (MongoDB\Driver\WriteConcern): Write concern.
 142       *
 143       * @param string  $databaseName   Database name
 144       * @param string  $collectionName Collection name
 145       * @param array[] $operations     List of write operations
 146       * @param array   $options        Command options
 147       * @throws InvalidArgumentException for parameter/option parsing errors
 148       */
 149      public function __construct($databaseName, $collectionName, array $operations, array $options = [])
 150      {
 151          if (empty($operations)) {
 152              throw new InvalidArgumentException('$operations is empty');
 153          }
 154  
 155          $expectedIndex = 0;
 156  
 157          foreach ($operations as $i => $operation) {
 158              if ($i !== $expectedIndex) {
 159                  throw new InvalidArgumentException(sprintf('$operations is not a list (unexpected index: "%s")', $i));
 160              }
 161  
 162              if (! is_array($operation)) {
 163                  throw InvalidArgumentException::invalidType(sprintf('$operations[%d]', $i), $operation, 'array');
 164              }
 165  
 166              if (count($operation) !== 1) {
 167                  throw new InvalidArgumentException(sprintf('Expected one element in $operation[%d], actually: %d', $i, count($operation)));
 168              }
 169  
 170              $type = key($operation);
 171              $args = current($operation);
 172  
 173              if (! isset($args[0]) && ! array_key_exists(0, $args)) {
 174                  throw new InvalidArgumentException(sprintf('Missing first argument for $operations[%d]["%s"]', $i, $type));
 175              }
 176  
 177              if (! is_array($args[0]) && ! is_object($args[0])) {
 178                  throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][0]', $i, $type), $args[0], 'array or object');
 179              }
 180  
 181              switch ($type) {
 182                  case self::INSERT_ONE:
 183                      break;
 184  
 185                  case self::DELETE_MANY:
 186                  case self::DELETE_ONE:
 187                      if (! isset($args[1])) {
 188                          $args[1] = [];
 189                      }
 190  
 191                      if (! is_array($args[1])) {
 192                          throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][1]', $i, $type), $args[1], 'array');
 193                      }
 194  
 195                      $args[1]['limit'] = ($type === self::DELETE_ONE ? 1 : 0);
 196  
 197                      if (isset($args[1]['collation'])) {
 198                          $this->isCollationUsed = true;
 199  
 200                          if (! is_array($args[1]['collation']) && ! is_object($args[1]['collation'])) {
 201                              throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][1]["collation"]', $i, $type), $args[1]['collation'], 'array or object');
 202                          }
 203                      }
 204  
 205                      $operations[$i][$type][1] = $args[1];
 206  
 207                      break;
 208  
 209                  case self::REPLACE_ONE:
 210                      if (! isset($args[1]) && ! array_key_exists(1, $args)) {
 211                          throw new InvalidArgumentException(sprintf('Missing second argument for $operations[%d]["%s"]', $i, $type));
 212                      }
 213  
 214                      if (! is_array($args[1]) && ! is_object($args[1])) {
 215                          throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][1]', $i, $type), $args[1], 'array or object');
 216                      }
 217  
 218                      if (is_first_key_operator($args[1])) {
 219                          throw new InvalidArgumentException(sprintf('First key in $operations[%d]["%s"][1] is an update operator', $i, $type));
 220                      }
 221  
 222                      if (! isset($args[2])) {
 223                          $args[2] = [];
 224                      }
 225  
 226                      if (! is_array($args[2])) {
 227                          throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][2]', $i, $type), $args[2], 'array');
 228                      }
 229  
 230                      $args[2]['multi'] = false;
 231                      $args[2] += ['upsert' => false];
 232  
 233                      if (isset($args[2]['collation'])) {
 234                          $this->isCollationUsed = true;
 235  
 236                          if (! is_array($args[2]['collation']) && ! is_object($args[2]['collation'])) {
 237                              throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][2]["collation"]', $i, $type), $args[2]['collation'], 'array or object');
 238                          }
 239                      }
 240  
 241                      if (! is_bool($args[2]['upsert'])) {
 242                          throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][2]["upsert"]', $i, $type), $args[2]['upsert'], 'boolean');
 243                      }
 244  
 245                      $operations[$i][$type][2] = $args[2];
 246  
 247                      break;
 248  
 249                  case self::UPDATE_MANY:
 250                  case self::UPDATE_ONE:
 251                      if (! isset($args[1]) && ! array_key_exists(1, $args)) {
 252                          throw new InvalidArgumentException(sprintf('Missing second argument for $operations[%d]["%s"]', $i, $type));
 253                      }
 254  
 255                      if (! is_array($args[1]) && ! is_object($args[1])) {
 256                          throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][1]', $i, $type), $args[1], 'array or object');
 257                      }
 258  
 259                      if (! is_first_key_operator($args[1]) && ! is_pipeline($args[1])) {
 260                          throw new InvalidArgumentException(sprintf('First key in $operations[%d]["%s"][1] is neither an update operator nor a pipeline', $i, $type));
 261                      }
 262  
 263                      if (! isset($args[2])) {
 264                          $args[2] = [];
 265                      }
 266  
 267                      if (! is_array($args[2])) {
 268                          throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][2]', $i, $type), $args[2], 'array');
 269                      }
 270  
 271                      $args[2]['multi'] = ($type === self::UPDATE_MANY);
 272                      $args[2] += ['upsert' => false];
 273  
 274                      if (isset($args[2]['arrayFilters'])) {
 275                          $this->isArrayFiltersUsed = true;
 276  
 277                          if (! is_array($args[2]['arrayFilters'])) {
 278                              throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][2]["arrayFilters"]', $i, $type), $args[2]['arrayFilters'], 'array');
 279                          }
 280                      }
 281  
 282                      if (isset($args[2]['collation'])) {
 283                          $this->isCollationUsed = true;
 284  
 285                          if (! is_array($args[2]['collation']) && ! is_object($args[2]['collation'])) {
 286                              throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][2]["collation"]', $i, $type), $args[2]['collation'], 'array or object');
 287                          }
 288                      }
 289  
 290                      if (! is_bool($args[2]['upsert'])) {
 291                          throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][2]["upsert"]', $i, $type), $args[2]['upsert'], 'boolean');
 292                      }
 293  
 294                      $operations[$i][$type][2] = $args[2];
 295  
 296                      break;
 297  
 298                  default:
 299                      throw new InvalidArgumentException(sprintf('Unknown operation type "%s" in $operations[%d]', $type, $i));
 300              }
 301  
 302              $expectedIndex += 1;
 303          }
 304  
 305          $options += ['ordered' => true];
 306  
 307          if (isset($options['bypassDocumentValidation']) && ! is_bool($options['bypassDocumentValidation'])) {
 308              throw InvalidArgumentException::invalidType('"bypassDocumentValidation" option', $options['bypassDocumentValidation'], 'boolean');
 309          }
 310  
 311          if (! is_bool($options['ordered'])) {
 312              throw InvalidArgumentException::invalidType('"ordered" option', $options['ordered'], 'boolean');
 313          }
 314  
 315          if (isset($options['session']) && ! $options['session'] instanceof Session) {
 316              throw InvalidArgumentException::invalidType('"session" option', $options['session'], Session::class);
 317          }
 318  
 319          if (isset($options['writeConcern']) && ! $options['writeConcern'] instanceof WriteConcern) {
 320              throw InvalidArgumentException::invalidType('"writeConcern" option', $options['writeConcern'], WriteConcern::class);
 321          }
 322  
 323          if (isset($options['writeConcern']) && $options['writeConcern']->isDefault()) {
 324              unset($options['writeConcern']);
 325          }
 326  
 327          $this->databaseName = (string) $databaseName;
 328          $this->collectionName = (string) $collectionName;
 329          $this->operations = $operations;
 330          $this->options = $options;
 331      }
 332  
 333      /**
 334       * Execute the operation.
 335       *
 336       * @see Executable::execute()
 337       * @param Server $server
 338       * @return BulkWriteResult
 339       * @throws UnsupportedException if array filters or collation is used and unsupported
 340       * @throws DriverRuntimeException for other driver errors (e.g. connection errors)
 341       */
 342      public function execute(Server $server)
 343      {
 344          if ($this->isArrayFiltersUsed && ! server_supports_feature($server, self::$wireVersionForArrayFilters)) {
 345              throw UnsupportedException::arrayFiltersNotSupported();
 346          }
 347  
 348          if ($this->isCollationUsed && ! server_supports_feature($server, self::$wireVersionForCollation)) {
 349              throw UnsupportedException::collationNotSupported();
 350          }
 351  
 352          $inTransaction = isset($this->options['session']) && $this->options['session']->isInTransaction();
 353          if ($inTransaction && isset($this->options['writeConcern'])) {
 354              throw UnsupportedException::writeConcernNotSupportedInTransaction();
 355          }
 356  
 357          $options = ['ordered' => $this->options['ordered']];
 358  
 359          if (! empty($this->options['bypassDocumentValidation']) &&
 360              server_supports_feature($server, self::$wireVersionForDocumentLevelValidation)
 361          ) {
 362              $options['bypassDocumentValidation'] = $this->options['bypassDocumentValidation'];
 363          }
 364  
 365          $bulk = new Bulk($options);
 366          $insertedIds = [];
 367  
 368          foreach ($this->operations as $i => $operation) {
 369              $type = key($operation);
 370              $args = current($operation);
 371  
 372              switch ($type) {
 373                  case self::DELETE_MANY:
 374                  case self::DELETE_ONE:
 375                      $bulk->delete($args[0], $args[1]);
 376                      break;
 377  
 378                  case self::INSERT_ONE:
 379                      $insertedIds[$i] = $bulk->insert($args[0]);
 380                      break;
 381  
 382                  case self::REPLACE_ONE:
 383                  case self::UPDATE_MANY:
 384                  case self::UPDATE_ONE:
 385                      $bulk->update($args[0], $args[1], $args[2]);
 386              }
 387          }
 388  
 389          $writeResult = $server->executeBulkWrite($this->databaseName . '.' . $this->collectionName, $bulk, $this->createOptions());
 390  
 391          return new BulkWriteResult($writeResult, $insertedIds);
 392      }
 393  
 394      /**
 395       * Create options for executing the bulk write.
 396       *
 397       * @see http://php.net/manual/en/mongodb-driver-server.executebulkwrite.php
 398       * @return array
 399       */
 400      private function createOptions()
 401      {
 402          $options = [];
 403  
 404          if (isset($this->options['session'])) {
 405              $options['session'] = $this->options['session'];
 406          }
 407  
 408          if (isset($this->options['writeConcern'])) {
 409              $options['writeConcern'] = $this->options['writeConcern'];
 410          }
 411  
 412          return $options;
 413      }
 414  }