Add definition rollup, action suggestions, and tables to the bucket report

Author: bean1352

modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts

@@ -29,7 +29,8 @@ import {
   MongoSyncBucketStorage,
   MongoSyncBucketStorageOptions,
   TopBucketCandidate,
-  TopBucketSelection
+  TopBucketSelection,
+  TopDefinitionCandidate
 } from '../MongoSyncBucketStorage.js';
 import {
   BucketDataDocumentV1,
@@ -190,13 +191,15 @@ export class MongoSyncBucketStorageV1 extends MongoSyncBucketStorage {
 
   // For storage v1/v2, bucket state and bucket data are shared collections scoped by group (replication stream).
   protected async collectTopBuckets(limit: number): Promise<TopBucketSelection> {
-    const { buckets, totals } = await this.aggregateTopBuckets(
+    const { buckets, definitions, definitionsTruncated, totals } = await this.aggregateTopBuckets(
       this.db.bucketStateV1,
       { '_id.g': this.replicationStreamId },
       limit
     );
     return {
       buckets: buckets.map((b) => ({ bucket: b.id.b, operations: b.operations, operationBytes: b.operationBytes })),
+      definitions,
+      definitionsTruncated,
       totals
     };
   }
@@ -225,6 +228,36 @@ export class MongoSyncBucketStorageV1 extends MongoSyncBucketStorage {
     return this.estimateRowsFromOperationSample(this.db.bucketDataV1, buildPrefix, candidate.operations, sampled);
   }
 
+  protected estimateDefinitionRows(candidate: TopDefinitionCandidate): Promise<BucketRowEstimate> {
+    const sampled = this.shouldSampleBucketRows(candidate.operations);
+    const buildPrefix = (applySample: boolean): mongo.Document[] => {
+      // All of a definition's bucket names start with `<definition>[`, so an `_id` range on that string
+      // prefix selects exactly the definition's operations via the index. `\\` (0x5C) is the character
+      // after `[` (0x5B), so [`<definition>[`, `<definition>\\`) cannot include any other definition:
+      // a longer definition name would have to differ at or before the `[`.
+      const prefix: mongo.Document[] = [
+        {
+          $match: {
+            _id: {
+              $gte: { g: this.replicationStreamId, b: `${candidate.definition}[`, o: new bson.MinKey() },
+              $lt: { g: this.replicationStreamId, b: `${candidate.definition}\\`, o: new bson.MinKey() }
+            }
+          }
+        }
+      ];
+      if (applySample) {
+        prefix.push({ $match: { $sampleRate: this.bucketRowSampleRate(candidate.operations) } });
+      }
+      return prefix;
+    };
+    // Include the bucket name in the row key: at definition grain a row counts once per bucket holding it.
+    return this.estimateRowsFromOperationSample(this.db.bucketDataV1, buildPrefix, candidate.operations, sampled, {
+      b: '$_id.b',
+      table: '$table',
+      row_id: '$row_id'
+    });
+  }
+
   protected createMongoParameterCompactor(
     checkpoint: InternalOpId,
     options: storage.CompactOptions

modules/module-mongodb-storage/src/storage/implementation/v1/MongoSyncBucketStorageV1.ts

@@ -28,7 +28,8 @@ import {
   MongoSyncBucketStorage,
   MongoSyncBucketStorageOptions,
   TopBucketCandidate,
-  TopBucketSelection
+  TopBucketSelection,
+  TopDefinitionCandidate
 } from '../MongoSyncBucketStorage.js';
 import { loadBucketDataDocument } from './bucket-format.js';
 import {
@@ -193,7 +194,7 @@ export class MongoSyncBucketStorageV3 extends MongoSyncBucketStorage {
   // sharing these collections. Scope to the active config's definition ids so the report excludes stale buckets
   // from old/stopped definitions. `this.storageIds` is derived from the active config only (see getActiveSyncConfig).
   protected async collectTopBuckets(limit: number): Promise<TopBucketSelection> {
-    const { buckets, totals } = await this.aggregateTopBuckets(
+    const { buckets, definitions, definitionsTruncated, totals } = await this.aggregateTopBuckets(
       this.db.bucketState(this.replicationStreamId),
       { '_id.d': { $in: this.storageIds.bucketDefinitionIds } },
       limit
@@ -205,6 +206,8 @@ export class MongoSyncBucketStorageV3 extends MongoSyncBucketStorage {
         operationBytes: b.operationBytes,
         defId: b.id.d
       })),
+      definitions,
+      definitionsTruncated,
       totals
     };
   }
@@ -229,6 +232,30 @@ export class MongoSyncBucketStorageV3 extends MongoSyncBucketStorage {
     return this.estimateRowsFromOperationSample(collection, buildPrefix, candidate.operations, sampled);
   }
 
+  protected estimateDefinitionRows(candidate: TopDefinitionCandidate): Promise<BucketRowEstimate> {
+    // A definition's operations are exactly its per-definition bucket_data collection, so no match stage is
+    // needed. Keep the bucket name alongside each unwound operation: at definition grain a row counts once
+    // per bucket holding it.
+    const sampled = this.shouldSampleBucketRows(candidate.operations);
+    const collection = this.db.bucketData(this.replicationStreamId, candidate.defId!);
+    const buildPrefix = (applySample: boolean): mongo.Document[] => {
+      const prefix: mongo.Document[] = [];
+      if (applySample) {
+        prefix.push({ $match: { $sampleRate: this.bucketRowSampleRate(candidate.operations) } });
+      }
+      prefix.push(
+        { $unwind: '$ops' },
+        { $project: { b: '$_id.b', op: '$ops.op', table: '$ops.table', row_id: '$ops.row_id' } }
+      );
+      return prefix;
+    };
+    return this.estimateRowsFromOperationSample(collection, buildPrefix, candidate.operations, sampled, {
+      b: '$b',
+      table: '$table',
+      row_id: '$row_id'
+    });
+  }
+
   protected createMongoParameterCompactor(
     checkpoint: InternalOpId,
     options: storage.CompactOptions

modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts

@@ -61,13 +61,33 @@ bucket_definitions:
     const report = await getReport(bucketStorage);
 
     expect(report.totals.bucketCount).toEqual(1);
-    expect(report.truncated).toEqual(false);
+    expect(report.bucketsTruncated).toEqual(false);
+    expect(report.definitionsTruncated).toEqual(false);
 
     const stats = report.buckets.find((b) => b.bucket === bucket)!;
     // Three inserts of distinct ids: three operations, three live rows, fully compacted (ratio 1).
-    expect(stats).toMatchObject({ operations: 3, rows: 3, fragmentation: 1, rowsEstimated: false });
+    expect(stats).toMatchObject({
+      operations: 3,
+      rows: 3,
+      fragmentation: 1,
+      rowsEstimated: false,
+      suggestedAction: 'none',
+      tables: ['test']
+    });
     expect(stats.operationBytes).toBeGreaterThan(0);
     expect(report.totals).toMatchObject({ operations: 3, estimated: false });
+
+    // The definition rollup aggregates the single bucket. The definition name is the bucket-name prefix.
+    expect(report.definitions).toHaveLength(1);
+    expect(report.definitions[0]).toMatchObject({
+      definition: bucket.split('[')[0],
+      bucketCount: 1,
+      operations: 3,
+      rows: 3,
+      fragmentation: 1,
+      suggestedAction: 'none',
+      tables: ['test']
+    });
   });
 
   test('operations exceed live rows after updates, and compaction reduces fragmentation', async () => {
@@ -162,6 +182,15 @@ bucket_definitions:
     expect(report.buckets.find((b) => b.bucket === b1)).toMatchObject({ operations: 3, rows: 1 });
     expect(report.buckets.find((b) => b.bucket === b2)).toMatchObject({ operations: 2, rows: 2 });
 
+    // Both buckets belong to one definition; the rollup sums them, counting each bucket's rows separately.
+    expect(report.definitions).toHaveLength(1);
+    expect(report.definitions[0]).toMatchObject({
+      definition: b1.split('[')[0],
+      bucketCount: 2,
+      operations: 5,
+      rows: 3
+    });
+
     // operationBytes is an aggregated ($toDouble) sum; assert every bucket is non-zero and that the
     // per-bucket bytes add up to the instance total.
     expect(report.totals.operationBytes).toBeGreaterThan(0);
@@ -204,13 +233,46 @@ bucket_definitions:
     const b1 = test_utils.bucketRequest(content, 'grouped["b1"]').bucket;
 
     const report = await getReport(bucketStorage, { limit: 1 });
-    expect(report.truncated).toEqual(true);
+    expect(report.bucketsTruncated).toEqual(true);
     expect(report.buckets.map((b) => b.bucket)).toEqual([b1]);
     // Totals still cover every bucket, not just the truncated list.
     expect(report.totals.bucketCount).toEqual(2);
     expect(report.totals).toMatchObject({ operations: 3, estimated: false });
   });
 
+  test('caps the definition rollup and flags the truncation', async () => {
+    // Two definitions past the rollup cap; a single row lands in every definition's global bucket.
+    const definitionCount = storage.BUCKET_REPORT_DEFINITION_LIMIT + 2;
+    const manyDefinitions =
+      'bucket_definitions:\n' +
+      Array.from({ length: definitionCount }, (_, i) => `  def${i}:\n    data: [select * from test]\n`).join('');
+
+    await using factory = await generateStorageFactory();
+    const { stream } = await test_utils.deploySyncRules(
+      factory,
+      updateSyncRulesFromYaml(manyDefinitions, { storageVersion })
+    );
+    const bucketStorage = factory.getInstance(stream);
+
+    await using writer = await bucketStorage.createWriter(test_utils.BATCH_OPTIONS);
+    const testTable = await test_utils.resolveTestTable(writer, 'test', ['id'], config);
+    await writer.markAllSnapshotDone('1/1');
+    await writer.save({
+      sourceTable: testTable,
+      tag: storage.SaveOperationTag.INSERT,
+      after: { id: 't1' },
+      afterReplicaId: test_utils.rid('t1')
+    });
+    await writer.commit('1/1');
+    await writer.flush();
+
+    const report = await getReport(bucketStorage);
+    expect(report.totals.bucketCount).toEqual(definitionCount);
+    expect(report.bucketsTruncated).toEqual(false);
+    expect(report.definitions).toHaveLength(storage.BUCKET_REPORT_DEFINITION_LIMIT);
+    expect(report.definitionsTruncated).toEqual(true);
+  });
+
   test('samples the row count for a bucket above the sampling threshold', async () => {
     await using factory = await generateStorageFactory();
     const { stream, content } = await test_utils.deploySyncRules(
@@ -265,5 +327,21 @@ bucket_definitions:
     expect(stats.rows).toBeLessThanOrEqual(55);
     // Fragmentation is operations / rows, so a heavily updated bucket reads well above 1.
     expect(stats.fragmentation).toBeGreaterThan(10);
+    // The history is un-compacted superseded churn, which a compact reclaims.
+    expect(stats.suggestedAction).toEqual('compact');
+    // The sampled history names the tables a defragment would touch.
+    expect(stats.tables).toEqual(['test']);
+
+    // The definition rollup samples the same history at definition grain.
+    expect(report.definitions).toHaveLength(1);
+    const defStats = report.definitions[0];
+    expect(defStats).toMatchObject({
+      bucketCount: 1,
+      operations: stats.operations,
+      suggestedAction: 'compact',
+      tables: ['test']
+    });
+    expect(defStats.rows).toBeGreaterThanOrEqual(45);
+    expect(defStats.rows).toBeLessThanOrEqual(55);
   });
 }

packages/service-core-tests/src/tests/register-bucket-report-tests.ts

@@ -313,15 +313,29 @@ export const bucketReport = routeDefinition({
         rows: bucket.rows,
         operation_bytes: bucket.operationBytes,
         fragmentation: bucket.fragmentation,
-        rows_estimated: bucket.rowsEstimated
+        rows_estimated: bucket.rowsEstimated,
+        suggested_action: bucket.suggestedAction,
+        tables: bucket.tables
+      })),
+      definitions: report.definitions.map((definition) => ({
+        definition: definition.definition,
+        bucket_count: definition.bucketCount,
+        operations: definition.operations,
+        operation_bytes: definition.operationBytes,
+        rows: definition.rows,
+        fragmentation: definition.fragmentation,
+        rows_estimated: definition.rowsEstimated,
+        suggested_action: definition.suggestedAction,
+        tables: definition.tables
       })),
       totals: {
         bucket_count: report.totals.bucketCount,
         operations: report.totals.operations,
         operation_bytes: report.totals.operationBytes,
         estimated: report.totals.estimated
       },
-      truncated: report.truncated
+      buckets_truncated: report.bucketsTruncated,
+      definitions_truncated: report.definitionsTruncated
     });
   }
 });