SERVER-63039 Add staged documents size target to BatchedDeleteStage

7 files changed, 89 insertions, 31 deletions

diff --git a/jstests/noPassthrough/batched_multi_deletes.js b/jstests/noPassthrough/batched_multi_deletes.js
index d77b5bb1b21..5daffa31778 100644
--- a/jstests/noPassthrough/batched_multi_deletes.js
+++ b/jstests/noPassthrough/batched_multi_deletes.js
@@ -51,8 +51,9 @@ function validateBatchedDeletes(conn) {
     assert.commandWorked(
         db.adminCommand({setParameter: 1, internalBatchUserMultiDeletesForTest: 1}));
 
-    // For consistent results, don't enforce the targetBatchTimeMS.
+    // For consistent results, don't enforce the targetBatchTimeMS and targetStagedDocBytes.
     assert.commandWorked(db.adminCommand({setParameter: 1, batchedDeletesTargetBatchTimeMS: 0}));
+    assert.commandWorked(db.adminCommand({setParameter: 1, batchedDeletesTargetStagedDocBytes: 0}));
 
     // Explain plan and executionStats.
     {
diff --git a/jstests/noPassthrough/batched_multi_deletes_failover.js b/jstests/noPassthrough/batched_multi_deletes_failover.js
index 7014f9f5b95..72afcd6c1c9 100644
--- a/jstests/noPassthrough/batched_multi_deletes_failover.js
+++ b/jstests/noPassthrough/batched_multi_deletes_failover.js
@@ -126,7 +126,7 @@ function runTest(failoverFn, clustered, expectNetworkErrorOnDelete) {
     assert.commandWorked(coll.createIndex({b: 1}));
 
     const hangAfterApproxNDocs = Random.randInt(collCount);
-    jsTestLog(`About to hang batched delete after evaluationg approximatly ${
+    jsTestLog(`About to hang batched delete after evaluating approximately ${
         hangAfterApproxNDocs} documents`);
 
     assert.commandWorked(
diff --git a/jstests/noPassthrough/batched_multi_deletes_oplog.js b/jstests/noPassthrough/batched_multi_deletes_oplog.js
index d1d83419ab7..313654f557b 100644
--- a/jstests/noPassthrough/batched_multi_deletes_oplog.js
+++ b/jstests/noPassthrough/batched_multi_deletes_oplog.js
@@ -23,9 +23,9 @@ function validateBatchedDeletesOplogDocsPerBatch(conn) {
 
     assert.commandWorked(
         db.adminCommand({setParameter: 1, internalBatchUserMultiDeletesForTest: 1}));
-    // Disable time-based batching
-    assert.commandWorked(db.adminCommand({setParameter: 1, batchedDeletesTargetBatchBytes: 0}));
     // Disable size-based batching
+    assert.commandWorked(db.adminCommand({setParameter: 1, batchedDeletesTargetStagedDocBytes: 0}));
+    // Disable time-based batching
     assert.commandWorked(db.adminCommand({setParameter: 1, batchedDeletesTargetBatchTimeMS: 0}));
     // Set docs per batch target
     assert.commandWorked(
@@ -66,9 +66,9 @@ function validateBatchedDeletesOplogBatchAbove16MB(conn) {
         Math.ceil(collCount / 63600 /* max docs per batch, see comment above. */);
     assert.commandWorked(
         db.adminCommand({setParameter: 1, internalBatchUserMultiDeletesForTest: 1}));
-    // Disable time-based batching
-    assert.commandWorked(db.adminCommand({setParameter: 1, batchedDeletesTargetBatchBytes: 0}));
     // Disable size-based batching
+    assert.commandWorked(db.adminCommand({setParameter: 1, batchedDeletesTargetStagedDocBytes: 0}));
+    // Disable time-based batching
     assert.commandWorked(db.adminCommand({setParameter: 1, batchedDeletesTargetBatchTimeMS: 0}));
     // Set artificially high docs per batch target
     assert.commandWorked(
diff --git a/jstests/noPassthrough/batched_multi_deletes_params.js b/jstests/noPassthrough/batched_multi_deletes_params.js
index cc0365bf728..b819f4de9a6 100644
--- a/jstests/noPassthrough/batched_multi_deletes_params.js
+++ b/jstests/noPassthrough/batched_multi_deletes_params.js
@@ -106,9 +106,51 @@ function validateTargetBatchTimeMS() {
     }
 }
 
+function validateTargetStagedDocsBytes() {
+    const collCount = 10000;
+    const docPaddingBytes = 1024;
+    const cumulativePaddingBytes = collCount *
+        (bsonsize({_id: ObjectId(), a: 'a'}) +
+         100 /* allow for getMemUsage() own metadata and overestimation */ + docPaddingBytes);
+
+    assert.commandWorked(db.adminCommand({setParameter: 1, batchedDeletesTargetBatchTimeMS: 0}));
+    assert.commandWorked(db.adminCommand({setParameter: 1, batchedDeletesTargetBatchDocs: 0}));
+
+    for (let stagedDocsBytes of [0, 1024 * 1024, 5 * 1024 * 1024]) {
+        jsTestLog("Validating stagedDocsBytes=" + stagedDocsBytes);
+
+        assert.commandWorked(db.adminCommand(
+            {setParameter: 1, batchedDeletesTargetStagedDocBytes: stagedDocsBytes}));
+
+        coll.drop();
+        assert.commandWorked(coll.insertMany(
+            [...Array(collCount).keys()].map(x => ({a: "a".repeat(docPaddingBytes)}))));
+
+        // batchedDeletesTargetStagedDocsBytes := 0 means no limit.
+        const expectedBatches =
+            stagedDocsBytes ? Math.ceil(cumulativePaddingBytes / stagedDocsBytes) : 1;
+        const serverStatusBatchesBefore = db.serverStatus()['batchedDeletes']['batches'];
+        const serverStatusDocsBefore = db.serverStatus()['batchedDeletes']['docs'];
+
+        assert.eq(collCount, coll.find().itcount());
+        assert.commandWorked(coll.deleteMany({}));
+        assert.eq(0, coll.find().itcount());
+
+        const serverStatusBatchesAfter = db.serverStatus()['batchedDeletes']['batches'];
+        const serverStatusDocsAfter = db.serverStatus()['batchedDeletes']['docs'];
+        const serverStatusDocsExpected = serverStatusDocsBefore + collCount;
+        const serverStatusBatchesExpected = serverStatusBatchesBefore + expectedBatches;
+        assert.eq(serverStatusBatchesAfter, serverStatusBatchesExpected);
+        assert.eq(serverStatusDocsAfter, serverStatusDocsExpected);
+
+        rst.awaitReplication();
+        rst.checkReplicatedDataHashes();
+    }
+}
+
 validateTargetDocsPerBatch();
 validateTargetBatchTimeMS();
-// TODO (SERVER-63039): validate targetStagedDocBytes.
+validateTargetStagedDocsBytes();
 
 rst.stopSet();
 })();
diff --git a/src/mongo/db/exec/batched_delete_stage.cpp b/src/mongo/db/exec/batched_delete_stage.cpp
index d04271f9c05..78bfd05e352 100644
--- a/src/mongo/db/exec/batched_delete_stage.cpp
+++ b/src/mongo/db/exec/batched_delete_stage.cpp
@@ -83,7 +83,11 @@ void incrementSSSMetricNoOverflow(AtomicWord<long long>& metric, long long value
  */
 struct BatchedDeletesSSS : ServerStatusSection {
     BatchedDeletesSSS()
-        : ServerStatusSection("batchedDeletes"), batches(0), docs(0), sizeBytes(0), timeMillis(0) {}
+        : ServerStatusSection("batchedDeletes"),
+          batches(0),
+          docs(0),
+          stagedSizeBytes(0),
+          timeMillis(0) {}
 
     bool includeByDefault() const override {
         return true;
@@ -93,7 +97,7 @@ struct BatchedDeletesSSS : ServerStatusSection {
         BSONObjBuilder bob;
         bob.appendNumber("batches", batches.loadRelaxed());
         bob.appendNumber("docs", docs.loadRelaxed());
-        bob.appendNumber("sizeBytes", sizeBytes.loadRelaxed());
+        bob.appendNumber("stagedSizeBytes", stagedSizeBytes.loadRelaxed());
         bob.append("timeMillis", timeMillis.loadRelaxed());
 
         return bob.obj();
@@ -101,7 +105,7 @@ struct BatchedDeletesSSS : ServerStatusSection {
 
     AtomicWord<long long> batches;
     AtomicWord<long long> docs;
-    AtomicWord<long long> sizeBytes;
+    AtomicWord<long long> stagedSizeBytes;
     AtomicWord<long long> timeMillis;
 } batchedDeletesSSS;
 
@@ -115,6 +119,7 @@ BatchedDeleteStage::BatchedDeleteStage(ExpressionContext* expCtx,
           kStageType.rawData(), expCtx, std::move(params), ws, collection, child),
       _batchParams(std::move(batchParams)),
       _stagedDeletesBuffer(ws),
+      _stagedDeletesWatermarkBytes(0),
       _drainRemainingBuffer(false) {
     tassert(6303800,
             "batched deletions only support multi-document deletions (multi: true)",
@@ -133,15 +138,9 @@ BatchedDeleteStage::BatchedDeleteStage(ExpressionContext* expCtx,
     tassert(6303805,
             "batched deletions do not support the 'numStatsForDoc' parameter",
             !_params->numStatsForDoc);
-    tassert(6303806,
-            "batch size cannot be unbounded; you must specify at least one of the following batch "
-            "parameters: "
-            "'targetBatchBytes', 'targetBatchDocs', 'targetBatchTimeMS'",
-            _batchParams->targetBatchBytes || _batchParams->targetBatchDocs ||
-                _batchParams->targetBatchTimeMS != Milliseconds(0));
     tassert(6303807,
             "batch size parameters must be greater than or equal to zero",
-            _batchParams->targetBatchBytes >= 0 && _batchParams->targetBatchDocs >= 0 &&
+            _batchParams->targetStagedDocBytes >= 0 && _batchParams->targetBatchDocs >= 0 &&
                 _batchParams->targetBatchTimeMS >= Milliseconds(0));
 }
 
@@ -248,7 +247,6 @@ PlanStage::StageState BatchedDeleteStage::_deleteBatch(WorkingSetID* out) {
     incrementSSSMetricNoOverflow(batchedDeletesSSS.docs, docsDeleted);
     incrementSSSMetricNoOverflow(batchedDeletesSSS.batches, 1);
     incrementSSSMetricNoOverflow(batchedDeletesSSS.timeMillis, batchTimer.millis());
-    // TODO (SERVER-63039): report batch size
     _specificStats.docsDeleted += docsDeleted;
 
     if (bufferOffset < _stagedDeletesBuffer.size()) {
@@ -322,10 +320,15 @@ PlanStage::StageState BatchedDeleteStage::doWork(WorkingSetID* out) {
             // retry deleting it.
             member->makeObjOwnedIfNeeded();
             _stagedDeletesBuffer.append(id);
+            const auto memberMemFootprintBytes = member->getMemUsage();
+            _stagedDeletesWatermarkBytes += memberMemFootprintBytes;
+            incrementSSSMetricNoOverflow(batchedDeletesSSS.stagedSizeBytes,
+                                         memberMemFootprintBytes);
         }
     }
 
     if (!_params->isExplain && (_drainRemainingBuffer || _batchTargetMet())) {
+        _stagedDeletesWatermarkBytes = 0;
         return _deleteBatch(out);
     }
 
@@ -355,8 +358,15 @@ void BatchedDeleteStage::_signalIfDrainComplete() {
 }
 
 bool BatchedDeleteStage::_batchTargetMet() {
-    return _batchParams->targetBatchDocs &&
-        _stagedDeletesBuffer.size() >=
-        static_cast<unsigned long long>(_batchParams->targetBatchDocs);
+    tassert(6303900,
+            "not expecting to be still draining staged deletions while evaluating whether to "
+            "commit staged deletions",
+            !_drainRemainingBuffer);
+    return (_batchParams->targetBatchDocs &&
+            _stagedDeletesBuffer.size() >=
+                static_cast<unsigned long long>(_batchParams->targetBatchDocs)) ||
+        (_batchParams->targetStagedDocBytes &&
+         _stagedDeletesWatermarkBytes >=
+             static_cast<unsigned long long>(_batchParams->targetStagedDocBytes));
 }
 }  // namespace mongo
diff --git a/src/mongo/db/exec/batched_delete_stage.h b/src/mongo/db/exec/batched_delete_stage.h
index 4aae7b0264b..6c8d859e97e 100644
--- a/src/mongo/db/exec/batched_delete_stage.h
+++ b/src/mongo/db/exec/batched_delete_stage.h
@@ -43,18 +43,18 @@ namespace mongo {
  */
 struct BatchedDeleteStageBatchParams {
     BatchedDeleteStageBatchParams()
-        : targetBatchBytes(gBatchedDeletesTargetBatchBytes.load()),
-          targetBatchDocs(gBatchedDeletesTargetBatchDocs.load()),
-          targetBatchTimeMS(Milliseconds(gBatchedDeletesTargetBatchTimeMS.load())) {}
+        : targetBatchDocs(gBatchedDeletesTargetBatchDocs.load()),
+          targetBatchTimeMS(Milliseconds(gBatchedDeletesTargetBatchTimeMS.load())),
+          targetStagedDocBytes(gBatchedDeletesTargetStagedDocBytes.load()) {}
 
-    // Documents staged for deletions are processed in a batch once this batch size target is met.
-    // Accounts for documents and indexes. A value of zero means unlimited.
-    long long targetBatchBytes = 0;
     // Documents staged for deletions are processed in a batch once this document count target is
     // met. A value of zero means unlimited.
     long long targetBatchDocs = 0;
     // A batch is committed as soon as this target execution time is met. Zero means unlimited.
     Milliseconds targetBatchTimeMS = Milliseconds(0);
+    // Documents staged for deletions are processed in a batch once this size target is met.
+    // Accounts for document size, not for indexes. A value of zero means unlimited.
+    long long targetStagedDocBytes = 0;
 };
 
 /**
@@ -115,6 +115,11 @@ private:
     // Holds information for each document staged for delete.
     BatchedDeleteStageBuffer _stagedDeletesBuffer;
 
+    // Holds the maximum cumulative size of all documents staged for delete. It is a watermark in
+    // that it resets to zero once the target is met and the staged documents start being processed,
+    // regardless of whether all staged deletes have been committed yet.
+    size_t _stagedDeletesWatermarkBytes;
+
     // Whether there are remaining docs in the buffer from a previous call to doWork() that should
     // be drained before fetching more documents.
     bool _drainRemainingBuffer;
diff --git a/src/mongo/db/exec/batched_delete_stage.idl b/src/mongo/db/exec/batched_delete_stage.idl
index 3794cc1836d..5ccf0cdab72 100644
--- a/src/mongo/db/exec/batched_delete_stage.idl
+++ b/src/mongo/db/exec/batched_delete_stage.idl
@@ -33,12 +33,12 @@ imports:
     - "mongo/idl/basic_types.idl"
 
 server_parameters:
-  batchedDeletesTargetBatchBytes:
-    description: "Threshold in bytes accounting for documents and index entries at which a batch of document deletions is committed. A value of zero means unlimited"
+  batchedDeletesTargetStagedDocBytes:
+    description: "Threshold in bytes accounting for documents (not index entries) at which a batch of document deletions is committed. A value of zero means unlimited"
     set_at: [startup, runtime]
     cpp_vartype: 'AtomicWord<long long>'
-    cpp_varname: gBatchedDeletesTargetBatchBytes
-    default: 52428800 # 50MB
+    cpp_varname: gBatchedDeletesTargetStagedDocBytes
+    default: 31457280 # 30MB
     validator:
       gte: 0
   batchedDeletesTargetBatchDocs: