returns a promise for a list of docs to update, based on the input docId. the order doesn't matter, because post-3.2.0, bulkDocs is an atomic operation in all three adapters.
501808
function getDocsToPersist(docId, view, docIdsToChangesAndEmits) {
  var metaDocId = '_local/doc_' + docId;
  var defaultMetaDoc = {_id: metaDocId, keys: []};
  var docData = docIdsToChangesAndEmits[docId];
  var indexableKeysToKeyValues = docData.indexableKeysToKeyValues;
  var changes = docData.changes;

  function getMetaDoc() {
    if (isGenOne(changes)) {
      // generation 1, so we can safely assume initial state
      // for performance reasons (avoids unnecessary GETs)
      return Promise.resolve(defaultMetaDoc);
    }
    return view.db.get(metaDocId).catch(defaultsTo(defaultMetaDoc));
  }

  function getKeyValueDocs(metaDoc) {
    if (!metaDoc.keys.length) {
      // no keys, no need for a lookup
      return Promise.resolve({rows: []});
    }
    return view.db.allDocs({
      keys: metaDoc.keys,
      include_docs: true
    });
  }

  function processKvDocs(metaDoc, kvDocsRes) {
    var kvDocs = [];
    var oldKeysMap = {};

    for (var i = 0, len = kvDocsRes.rows.length; i < len; i++) {
      var row = kvDocsRes.rows[i];
      var doc = row.doc;
      if (!doc) { // deleted
        continue;
      }
      kvDocs.push(doc);
      oldKeysMap[doc._id] = true;
      doc._deleted = !indexableKeysToKeyValues[doc._id];
      if (!doc._deleted) {
        var keyValue = indexableKeysToKeyValues[doc._id];
        if ('value' in keyValue) {
          doc.value = keyValue.value;
        }
      }
    }

    var newKeys = Object.keys(indexableKeysToKeyValues);
    newKeys.forEach(function (key) {
      if (!oldKeysMap[key]) {
        // new doc
        var kvDoc = {
          _id: key
        };
        var keyValue = indexableKeysToKeyValues[key];
        if ('value' in keyValue) {
          kvDoc.value = keyValue.value;
        }
        kvDocs.push(kvDoc);
      }
    });
    metaDoc.keys = uniq(newKeys.concat(metaDoc.keys));
    kvDocs.push(metaDoc);

    return kvDocs;
  }

  return getMetaDoc().then(function (metaDoc) {
    return getKeyValueDocs(metaDoc).then(function (kvDocsRes) {
      return processKvDocs(metaDoc, kvDocsRes);
    });
  });
}
511702
function (db, id, doc, metadata, cb) {
			var operations = [
				{
					Method: "PUT",
					Document: doc,
					Metadata: metadata,
					Key: id
				}
			];
			request.post({
				url: host + 'databases/' + db + '/bulk_docs',
				headers: {
					'Content-Type': 'application/json; charset=utf-8'
				},
				body: JSON.stringify(operations)
			}, function (error, response, resBody) {
				if (!error && response.statusCode === 200) {
					var result = JSON.parse(resBody);
					cb(null, result);
				} else {
					cb(error || new Error(response.statusCode), null);
				}
			});
		}
1348284
function upsert(db, docId, diffFun) {
  return new Promise(function (fulfill, reject) {
    db.get(docId, function (err, doc) {
      if (err) {
        /* istanbul ignore next */
        if (err.status !== 404) {
          return reject(err);
        }
        doc = {};
      }

      // the user might change the _rev, so save it for posterity
      var docRev = doc._rev;
      var newDoc = diffFun(doc);

      if (!newDoc) {
        // if the diffFun returns falsy, we short-circuit as
        // an optimization
        return fulfill({updated: false, rev: docRev});
      }

      // users aren't allowed to modify these values,
      // so reset them here
      newDoc._id = docId;
      newDoc._rev = docRev;
      fulfill(tryAndPut(db, newDoc, diffFun));
    });
  });
}
685477
function executeBatch(bulkOperation, batch, options, callback) {
  function resultHandler(err, result) {
    // Error is a driver related error not a bulk op error, terminate
    if (((err && err.driver) || (err && err.message)) && !(err instanceof MongoWriteConcernError)) {
      return handleCallback(callback, err);
    }

    // If we have and error
    if (err) err.ok = 0;
    if (err instanceof MongoWriteConcernError) {
      return handleMongoWriteConcernError(batch, bulkOperation.s.bulkResult, false, err, callback);
    }
    handleCallback(
      callback,
      null,
      mergeBatchResults(false, batch, bulkOperation.s.bulkResult, err, result)
    );
  }

  bulkOperation.finalOptionsHandler({ options, batch, resultHandler }, callback);
}
939132
function getDocument(document, docdbClient, nodeContext) {
    var documentUrl = `${collectionUrl}/docs/${document.id}`;
    nodeContext.log(`Getting document:\n${document.id}\n`);

    return new Promise((resolve, reject) => {
        nodeContext.log("trying read");
        docdbClient.readDocument(documentUrl, (err, result) => {
            nodeContext.log("reading");
            if (err) {
                nodeContext.log("error");
                if (err.code == HttpStatusCodes.NOTFOUND) {
                    nodeContext.log("creating");
                    docdbClient.createDocument(collectionUrl, document, (err, created) => {
                        nodeContext.log("try to create");
                        if (err) reject(err)
                        else resolve(created);
                    });
                } else {
                    reject(err);
                }
            } else {
                resolve(result);
            }
        });
    });
}
500477
function(db, couchdb, concurrentDocLimit, docsToDownload) {
  if (docsToDownload.length) {
    var changeSet = docsToDownload.splice(0, concurrentDocLimit);

    return couchdb.allDocs({
      include_docs: true,
      keys: _.pluck(changeSet, 'id')
    }).then(function(couchDbResult) {
      log.debug('Pulled ' + couchDbResult.rows.length + ' results from couchdb');

      log.debug('Clearing any existing documents from postgresql');

      return deleteDocuments(db, _.pluck(couchDbResult.rows, 'id'))
        .then(function() {
          return couchDbResult;
        });
    }).then(function(couchDbResult) {
      log.debug('Inserting ' + couchDbResult.rows.length + ' results into postgresql');

      var insertSql = format(INSERT_DOC_STMT,
        couchDbResult.rows.map(function(row) {
          return [row.doc];
        }));

      insertSql = sanitise(insertSql);

      return db.query(insertSql);
    }).then(function() {
      return loadAndStoreDocs(db, couchdb, concurrentDocLimit, docsToDownload);
    });
  }
}
1080223
function addToOperationsList(bulkOperation, docType, document) {
  // Get the bsonSize
  const bsonSize = bson.calculateObjectSize(document, {
    checkKeys: false
  });

  // Throw error if the doc is bigger than the max BSON size
  if (bsonSize >= bulkOperation.s.maxBatchSizeBytes)
    throw toError('document is larger than the maximum size ' + bulkOperation.s.maxBatchSizeBytes);

  // Create a new batch object if we don't have a current one
  if (bulkOperation.s.currentBatch == null)
    bulkOperation.s.currentBatch = new Batch(docType, bulkOperation.s.currentIndex);

  // Check if we need to create a new batch
  if (
    bulkOperation.s.currentBatchSize + 1 >= bulkOperation.s.maxWriteBatchSize ||
    bulkOperation.s.currentBatchSizeBytes + bulkOperation.s.currentBatchSizeBytes >=
      bulkOperation.s.maxBatchSizeBytes ||
    bulkOperation.s.currentBatch.batchType !== docType
  ) {
    // Save the batch to the execution stack
    bulkOperation.s.batches.push(bulkOperation.s.currentBatch);

    // Create a new batch
    bulkOperation.s.currentBatch = new Batch(docType, bulkOperation.s.currentIndex);

    // Reset the current size trackers
    bulkOperation.s.currentBatchSize = 0;
    bulkOperation.s.currentBatchSizeBytes = 0;
  } else {
    // Update current batch size
    bulkOperation.s.currentBatchSize = bulkOperation.s.currentBatchSize + 1;
    bulkOperation.s.currentBatchSizeBytes = bulkOperation.s.currentBatchSizeBytes + bsonSize;
  }

  if (docType === common.INSERT) {
    bulkOperation.s.bulkResult.insertedIds.push({
      index: bulkOperation.s.currentIndex,
      _id: document._id
    });
  }

  // We have an array of documents
  if (Array.isArray(document)) {
    throw toError('operation passed in cannot be an Array');
  } else {
    bulkOperation.s.currentBatch.originalIndexes.push(bulkOperation.s.currentIndex);
    bulkOperation.s.currentBatch.operations.push(document);
    bulkOperation.s.currentBatchSizeBytes = bulkOperation.s.currentBatchSizeBytes + bsonSize;
    bulkOperation.s.currentIndex = bulkOperation.s.currentIndex + 1;
  }

  // Return bulkOperation
  return bulkOperation;
}
844453
def bulk_load_docs(es, docs):
    

    chunk_size = 200

    try:
        results = elasticsearch.helpers.bulk(es, docs, chunk_size=chunk_size)
        log.debug(f"Elasticsearch documents loaded: {results[0]}")

        # elasticsearch.helpers.parallel_bulk(es, terms, chunk_size=chunk_size, thread_count=4)
        if len(results[1]) > 0:
            log.error("Bulk load errors {}".format(results))
    except elasticsearch.ElasticsearchException as e:
        log.error("Indexing error: {}\n".format(e))
463061
protected void pullBulkRevisions(List<RevisionInternal> bulkRevs) {

        int nRevs = bulkRevs.size();
        if (nRevs == 0) {
            return;
        }
        Log.d(TAG, "%s bulk-fetching %d remote revisions...", this, nRevs);
        Log.d(TAG, "%s bulk-fetching remote revisions: %s", this, bulkRevs);

        if (!canBulkGet) {
            pullBulkWithAllDocs(bulkRevs);
            return;
        }

        Log.v(TAG, "%s: POST _bulk_get", this);
        final List<RevisionInternal> remainingRevs = new ArrayList<RevisionInternal>(bulkRevs);

        ++httpConnectionCount;

        final RemoteBulkDownloaderRequest downloader;
        try {
            downloader = new RemoteBulkDownloaderRequest(
                    clientFactory,
                    remote,
                    true,
                    bulkRevs,
                    db,
                    this.requestHeaders,
                    new RemoteBulkDownloaderRequest.BulkDownloaderDocument() {
                        public void onDocument(Map<String, Object> props, long size) {
                            // Got a revision!
                            // Find the matching revision in 'remainingRevs' and get its sequence:
                            RevisionInternal rev;
                            if (props.get("_id") != null) {
                                rev = new RevisionInternal(props, size);
                            } else {
                                rev = new RevisionInternal((String) props.get("id"),
                                        (String) props.get("rev"), false);
                            }

                            int pos = remainingRevs.indexOf(rev);
                            if (pos > -1) {
                                rev.setSequence(remainingRevs.get(pos).getSequence());
                                remainingRevs.remove(pos);
                            } else {
                                Log.w(TAG, "%s : Received unexpected rev rev", this);
                            }

                            if (props.get("_id") != null) {
                                // Add to batcher ... eventually it will be fed to -insertRevisions:.
                                queueDownloadedRevision(rev);
                            } else {
                                Status status = statusFromBulkDocsResponseItem(props);
                                Throwable err = new CouchbaseLiteException(status);
                                revisionFailed(rev, err);
                            }
                        }
                    },
                    new RemoteRequestCompletion() {
                        public void onCompletion(RemoteRequest remoteRequest, Response httpResponse, Object result, Throwable e) {
                            // The entire _bulk_get is finished:
                            if (e != null) {
                                setError(e);
                                completedChangesCount.addAndGet(remainingRevs.size());
                            }
                            --httpConnectionCount;
                            // Start another task if there are still revisions waiting to be pulled:
                            pullRemoteRevisions();

                            if (cancellables != null && cancellables.values() != null && remoteRequest != null)
                                cancellables.values().remove(remoteRequest);
                        }
                    }
            );
        } catch (Exception e) {
            Log.e(TAG, "%s: pullBulkRevisions Exception: %s", this, e);
            return;
        }

        downloader.setAuthenticator(getAuthenticator());
        // set compressed request - gzip
        downloader.setCompressedRequest(canSendCompressedRequests());

        synchronized (remoteRequestExecutor) {
            if (!remoteRequestExecutor.isShutdown()) {
                Future future = remoteRequestExecutor.submit(downloader);
                pendingFutures.add(future);
                cancellables.put(future, downloader);
            }
        }
    }
571635
def _insert(self, docs, ordered=True, check_keys=True,
                manipulate=False, write_concern=None, op_id=None,
                bypass_doc_val=False, session=None):
        """"""
        if isinstance(docs, abc.Mapping):
            return self._insert_one(
                docs, ordered, check_keys, manipulate, write_concern, op_id,
                bypass_doc_val, session)

        ids = []

        if manipulate:
            def gen():
                """Generator that applies SON manipulators to each document
                and adds _id if necessary.
                """
                _db = self.__database
                for doc in docs:
                    # Apply user-configured SON manipulators. This order of
                    # operations is required for backwards compatibility,
                    # see PYTHON-709.
                    doc = _db._apply_incoming_manipulators(doc, self)
                    if not (isinstance(doc, RawBSONDocument) or '_id' in doc):
                        doc['_id'] = ObjectId()

                    doc = _db._apply_incoming_copying_manipulators(doc, self)
                    ids.append(doc['_id'])
                    yield doc
        else:
            def gen():
                """Generator that only tracks existing _ids."""
                for doc in docs:
                    # Don't inflate RawBSONDocument by touching fields.
                    if not isinstance(doc, RawBSONDocument):
                        ids.append(doc.get('_id'))
                    yield doc

        write_concern = write_concern or self._write_concern_for(session)
        blk = _Bulk(self, ordered, bypass_doc_val)
        blk.ops = [(message._INSERT, doc) for doc in gen()]
        try:
            blk.execute(write_concern, session=session)
        except BulkWriteError as bwe:
            _raise_last_error(bwe.details)
        return ids
1264597
function getDocs(fileName) {
        if (!promiseCache[fileName]) {
            var result = new $.Deferred();

            var path = ExtensionUtils.getModulePath(module, fileName),
                file = FileSystem.getFileForPath(path);

            FileUtils.readAsText(file)
                .done(function (text) {
                    var jsonData;
                    try {
                        jsonData = JSON.parse(text);
                    } catch (ex) {
                        console.error("Malformed documentation database: ", ex);
                        result.reject();
                    }
                    result.resolve(jsonData);  // ignored if we already reject()ed above
                })
                .fail(function (err) {
                    console.error("Unable to load documentation database: ", err);
                    result.reject();
                });

            promiseCache[fileName] = result.promise();
        }

        return promiseCache[fileName];
    }
933083
ApiFuture<List<DocumentSnapshot>> getAll(
      final DocumentReference[] documentReferences,
      @Nullable FieldMask fieldMask,
      @Nullable ByteString transactionId) {
    final SettableApiFuture<List<DocumentSnapshot>> futureList = SettableApiFuture.create();
    final Map<DocumentReference, DocumentSnapshot> resultMap = new HashMap<>();

    ApiStreamObserver<BatchGetDocumentsResponse> responseObserver =
        new ApiStreamObserver<BatchGetDocumentsResponse>() {
          int numResponses;

          @Override
          public void onNext(BatchGetDocumentsResponse response) {
            DocumentReference documentReference;
            DocumentSnapshot documentSnapshot;

            numResponses++;
            if (numResponses == 1) {
              tracer.getCurrentSpan().addAnnotation("Firestore.BatchGet: First response");
            } else if (numResponses % 100 == 0) {
              tracer.getCurrentSpan().addAnnotation("Firestore.BatchGet: Received 100 responses");
            }

            switch (response.getResultCase()) {
              case FOUND:
                documentReference =
                    new DocumentReference(
                        FirestoreImpl.this, ResourcePath.create(response.getFound().getName()));
                documentSnapshot =
                    DocumentSnapshot.fromDocument(
                        FirestoreImpl.this,
                        Timestamp.fromProto(response.getReadTime()),
                        response.getFound());
                break;
              case MISSING:
                documentReference =
                    new DocumentReference(
                        FirestoreImpl.this, ResourcePath.create(response.getMissing()));
                documentSnapshot =
                    DocumentSnapshot.fromMissing(
                        FirestoreImpl.this,
                        documentReference,
                        Timestamp.fromProto(response.getReadTime()));
                break;
              default:
                return;
            }

            resultMap.put(documentReference, documentSnapshot);
          }

          @Override
          public void onError(Throwable throwable) {
            tracer.getCurrentSpan().addAnnotation("Firestore.BatchGet: Error");
            futureList.setException(throwable);
          }

          @Override
          public void onCompleted() {
            tracer.getCurrentSpan().addAnnotation("Firestore.BatchGet: Complete");
            List<DocumentSnapshot> documentSnapshots = new ArrayList<>();

            for (DocumentReference documentReference : documentReferences) {
              documentSnapshots.add(resultMap.get(documentReference));
            }

            futureList.set(documentSnapshots);
          }
        };

    BatchGetDocumentsRequest.Builder request = BatchGetDocumentsRequest.newBuilder();
    request.setDatabase(getDatabaseName());

    if (fieldMask != null) {
      request.setMask(fieldMask.toPb());
    }

    if (transactionId != null) {
      request.setTransaction(transactionId);
    }

    for (DocumentReference docRef : documentReferences) {
      request.addDocuments(docRef.getName());
    }

    tracer
        .getCurrentSpan()
        .addAnnotation(
            "Firestore.BatchGet: Start",
            ImmutableMap.of(
                "numDocuments", AttributeValue.longAttributeValue(documentReferences.length)));

    streamRequest(request.build(), responseObserver, firestoreClient.batchGetDocumentsCallable());

    return futureList;
  }
552312
def commit(self, force=False):
        
        self.log.debug('Bulk commit requested')
        size = sys.getsizeof(self.docs)

        self.log.debug('Size of docs in KB: %d', size)
        if size > self.commit_threshold or force:
            self.log.info('Commiting %d KB to CouchDB' % size)
            self.my_db.update(self.docs)
            self.docs = []
641816
def get_async_doc(id, opts = {})
      data, _status_code, _headers = get_async_doc_with_http_info(id, opts)
      return data
    end
1313035
public Map<String, Object> getDocument(String collectionName, String docId) {
        Map<String, Map<String,Object>> docs = DDPStateSingleton.getInstance()
                .getCollection(collectionName);
        if (docs != null) {
            return docs.get(docId);
        }
        return null;
    }
239149
function push (state, docsOrIds) {
  var pushedObjects = []
  var errors = state.db.constructor.Errors

  return Promise.resolve(state.remote)

  .then(function (remote) {
    return new Promise(function (resolve, reject) {
      if (Array.isArray(docsOrIds)) {
        docsOrIds = docsOrIds.map(toId)
      } else {
        docsOrIds = docsOrIds && [toId(docsOrIds)]
      }

      if (docsOrIds && docsOrIds.filter(Boolean).length !== docsOrIds.length) {
        return Promise.reject(errors.NOT_AN_OBJECT)
      }

      var replication = state.db.replicate.to(remote, {
        create_target: true,
        doc_ids: docsOrIds,
        include_docs: true
      })

      /* istanbul ignore next */
      replication.catch(function () {
        // handled trough 'error' event
      })

      replication.on('complete', function () {
        resolve(pushedObjects)
      })
      replication.on('error', reject)
      replication.on('change', function (change) {
        pushedObjects = pushedObjects.concat(change.docs)

        for (var i = 0; i < change.docs.length; i++) {
          state.emitter.emit('push', change.docs[i])
        }
      })
    })
  })
}
349645
def update(collection_name, upsert, multi, spec, doc, safe, last_error_args):
    
    options = 0
    if upsert:
        options += 1
    if multi:
        options += 2

    data = __ZERO
    data += bson._make_c_string(collection_name)
    data += struct.pack("<i", options)
    data += bson.BSON.encode(spec)
    data += bson.BSON.encode(doc)
    if safe:
        (_, update_message) = __pack_message(2001, data)
        (request_id, error_message) = __last_error(last_error_args)
        return (request_id, update_message + error_message)
    else:
        return __pack_message(2001, data)