[MongoDB] Direct BSON Buffer -> JSON conversion (#599)

* Add experimental bufferToSqlite implementation.

* Use shared writer to reduce allocations.

* Add bson -> SqliteRow tests.

* Fix issues picked up by tests.

* Restructure tests to include the actual expected output.

* Fix handling of UUID edge cases and non-finite numbers.

* Further restructure and simplify tests.

* Improve and test Regexp option handling.

* Further regexp tests.

* Add tests for invalid UTF-8.

* Add some length checks for invalid BSON.

* Add tests for degenerate arrays.

* Initial restructuring and comments.

* Use consts.

* Further cleanup and comments.

* Simplify date serialization.

* Handle date options in compatibility context.

* Update docs.

* Rename converters, switch to the new one.

* Return Uint8Array instead of Buffer.

* Use raw queries for mongo_test.

* Remove DBRef tests - those were not actually using DBPointer types.

* Add more DBPointer and DBRef tests.

* Fix benchmark script; tweak docs.

* Fix DBPointer implementation.

* Avoid parsing a string just to get the end position.

* Custom parser for parseDocumentId.

* Optimize UUID implementation.

* Add surrogate pair tests.

Author: rkistner

modules/module-mongodb/scripts/benchmark-change-document-json.mts

@@ -0,0 +1,358 @@
+import { CompatibilityContext } from '@powersync/service-sync-rules';
+import * as bson from 'bson';
+import { performance } from 'node:perf_hooks';
+import { DirectSourceRowConverter, LegacySourceRowConverter, parseChangeDocument } from '../dist/index.js';
+
+// This is a synthetic benchmark to test performance of parseChangeDocument
+// versus the normal bson.deserialize().
+// Primarily AI-generated.
+
+const BSON_OPTIONS = { useBigInt64: true } as const;
+const OPERATION_TYPES = ['insert', 'update'] as const;
+const SIZE_TARGETS = [
+  { label: '1 KB', bytes: 1_024 },
+  { label: '10 KB', bytes: 10_240 },
+  { label: '100 KB', bytes: 102_400 }
+] as const;
+
+type OperationType = (typeof OPERATION_TYPES)[number];
+
+type FullDocument = {
+  _id: bson.ObjectId;
+  checksum: number;
+  operationType: OperationType;
+  tenantId: string;
+  version: number;
+  createdAt: Date;
+  updatedAt: Date;
+  flags: {
+    active: boolean;
+    archived: boolean;
+    source: string;
+  };
+  metrics: {
+    itemCount: number;
+    ratio: number;
+  };
+  tags: string[];
+  nested: {
+    owner: {
+      id: string;
+      region: string;
+    };
+    counters: number[];
+    changedFields: string[];
+  };
+  payload: string;
+};
+
+type UpdateDescription = {
+  updatedFields: {
+    payload: string;
+    metrics: FullDocument['metrics'];
+    updatedAt: Date;
+    version: number;
+  };
+  removedFields: string[];
+  truncatedArrays: never[];
+};
+
+type ChangeDocument = {
+  _id: {
+    _data: string;
+  };
+  operationType: OperationType;
+  wallTime: Date;
+  ns: {
+    db: string;
+    coll: string;
+  };
+  lsid: {
+    id: bson.Binary;
+  };
+  txnNumber: bson.Long;
+  documentKey: {
+    _id: bson.ObjectId;
+  };
+  fullDocument: FullDocument;
+  updateDescription?: UpdateDescription;
+};
+
+type Benchmark = {
+  label: string;
+  run: (buffer: Buffer) => number;
+};
+
+type Scenario = {
+  label: string;
+  operationType: OperationType;
+  targetBytes: number;
+  fullDocumentBytes: number;
+  eventBytes: number;
+  buffer: Buffer;
+};
+
+type BenchmarkResult = {
+  elapsedMs: number;
+  opsPerSecond: number;
+  mibPerSecond: number;
+};
+
+const rawConverter = new DirectSourceRowConverter(CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY);
+const docConverter = new LegacySourceRowConverter(CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY);
+
+const BENCHMARKS: readonly Benchmark[] = [
+  {
+    label: 'parseChangeDocument + rawToSqliteRow',
+    run: (buffer: Buffer) => {
+      const change = parseChangeDocument(buffer);
+      if (!('fullDocument' in change)) {
+        throw new Error('Unsupported change type: ' + change.operationType);
+      }
+      if (change.fullDocument == null) {
+        throw new Error('Expected fullDocument to be present');
+      }
+      const fullDocument = rawConverter.rawToSqliteRow(change.fullDocument) as { checksum?: number };
+      return Number(fullDocument.checksum ?? 0);
+    }
+  },
+  {
+    label: 'bson.deserialize + documentToSqliteRow',
+    run: (buffer: Buffer) => {
+      const change = bson.deserialize(buffer, BSON_OPTIONS) as { fullDocument?: { checksum?: number } };
+      const fullDocument = docConverter.documentToSqliteRow(change.fullDocument!);
+      return Number(change.fullDocument?.checksum ?? 0);
+    }
+  }
+] as const;
+
+function createFullDocument(operationType: OperationType, targetBytes: number): FullDocument {
+  const seed = targetBytes + (operationType === 'insert' ? 11 : 29);
+  const baseDocument: FullDocument = {
+    _id: new bson.ObjectId(),
+    checksum: seed,
+    operationType,
+    tenantId: 'tenant-benchmark',
+    version: operationType === 'insert' ? 1 : 2,
+    createdAt: new Date('2026-01-01T00:00:00.000Z'),
+    updatedAt: new Date('2026-01-02T03:04:05.000Z'),
+    flags: {
+      active: true,
+      archived: false,
+      source: 'benchmark'
+    },
+    metrics: {
+      itemCount: seed,
+      ratio: Number((targetBytes / 1024).toFixed(3))
+    },
+    tags: ['alpha', 'beta', 'gamma', operationType],
+    nested: {
+      owner: {
+        id: `owner-${seed}`,
+        region: 'af-south-1'
+      },
+      counters: [1, 2, 3, 5, 8, 13],
+      changedFields: operationType === 'update' ? ['payload', 'metrics.itemCount', 'updatedAt'] : []
+    },
+    payload: ''
+  };
+
+  const payloadLength = findPayloadLength(baseDocument, targetBytes);
+  const payload = repeatCharacter('x', payloadLength);
+  return {
+    ...baseDocument,
+    payload
+  };
+}
+
+function createChangeDocument(fullDocument: FullDocument, operationType: OperationType): ChangeDocument {
+  const updateDescription: UpdateDescription | undefined =
+    operationType === 'update'
+      ? {
+          updatedFields: {
+            payload: fullDocument.payload,
+            metrics: fullDocument.metrics,
+            updatedAt: fullDocument.updatedAt,
+            version: fullDocument.version
+          },
+          removedFields: ['legacyField'],
+          truncatedArrays: []
+        }
+      : undefined;
+
+  return {
+    _id: {
+      _data: `${operationType}-${fullDocument.checksum}-${new bson.ObjectId().toHexString()}`
+    },
+    operationType,
+    wallTime: new Date('2026-01-03T09:10:11.000Z'),
+    ns: {
+      db: 'benchmark_db',
+      coll: 'benchmark_coll'
+    },
+    lsid: {
+      id: new bson.Binary(Buffer.alloc(16, operationType === 'insert' ? 0x11 : 0x22))
+    },
+    txnNumber: seedLong(fullDocument.checksum),
+    documentKey: {
+      _id: fullDocument._id
+    },
+    ...(updateDescription == null ? {} : { updateDescription }),
+    fullDocument
+  };
+}
+
+function seedLong(value: number): bson.Long {
+  return bson.Long.fromNumber(value);
+}
+
+function findPayloadLength(baseDocument: FullDocument, targetBytes: number): number {
+  const baseSize = bson.calculateObjectSize(baseDocument);
+  if (baseSize >= targetBytes) {
+    return 0;
+  }
+
+  let low = 0;
+  let high = Math.max(16, targetBytes - baseSize);
+  while (calculateSizedDocumentBytes(baseDocument, high) < targetBytes) {
+    high *= 2;
+  }
+
+  while (low < high) {
+    const mid = Math.floor((low + high) / 2);
+    if (calculateSizedDocumentBytes(baseDocument, mid) < targetBytes) {
+      low = mid + 1;
+    } else {
+      high = mid;
+    }
+  }
+
+  return low;
+}
+
+function calculateSizedDocumentBytes(baseDocument: FullDocument, payloadLength: number): number {
+  return bson.calculateObjectSize({
+    ...baseDocument,
+    payload: repeatCharacter('x', payloadLength)
+  });
+}
+
+function repeatCharacter(character: string, count: number): string {
+  return character.repeat(Math.max(0, count));
+}
+
+function chooseIterations(eventBytes: number): number {
+  const targetBytes = 256 * 1024 * 1024;
+  return clamp(Math.floor(targetBytes / eventBytes), 200, 50_000);
+}
+
+function clamp(value: number, min: number, max: number): number {
+  return Math.max(min, Math.min(max, value));
+}
+
+function median(values: number[]): number {
+  const sorted = [...values].sort((a, b) => a - b);
+  const middle = Math.floor(sorted.length / 2);
+  return sorted.length % 2 === 0 ? (sorted[middle - 1] + sorted[middle]) / 2 : sorted[middle];
+}
+
+function buildScenario(operationType: OperationType, sizeLabel: string, targetBytes: number): Scenario {
+  const fullDocument = createFullDocument(operationType, targetBytes);
+  const changeDocument = createChangeDocument(fullDocument, operationType);
+  const buffer = Buffer.from(bson.serialize(changeDocument));
+  return {
+    label: `${operationType} ${sizeLabel}`,
+    operationType,
+    targetBytes,
+    fullDocumentBytes: bson.calculateObjectSize(fullDocument),
+    eventBytes: buffer.byteLength,
+    buffer
+  };
+}
+
+function runBenchmark(
+  label: string,
+  fn: (buffer: Buffer) => number,
+  buffer: Buffer,
+  iterations: number
+): BenchmarkResult {
+  const warmupIterations = Math.min(2_000, Math.max(50, Math.floor(iterations / 10)));
+  let sink = 0;
+  for (let i = 0; i < warmupIterations; i += 1) {
+    sink += fn(buffer);
+  }
+
+  const samples: number[] = [];
+  for (let round = 0; round < 5; round += 1) {
+    const start = performance.now();
+    for (let i = 0; i < iterations; i += 1) {
+      sink += fn(buffer);
+    }
+    samples.push(performance.now() - start);
+  }
+
+  if (sink === Number.MIN_SAFE_INTEGER) {
+    console.error(label);
+  }
+
+  const elapsedMs = median(samples);
+  const opsPerSecond = (iterations * 1000) / elapsedMs;
+  const mibPerSecond = (buffer.byteLength * iterations) / (1024 * 1024) / (elapsedMs / 1000);
+  return {
+    elapsedMs,
+    opsPerSecond,
+    mibPerSecond
+  };
+}
+
+function printRow(values: string[]): void {
+  const widths = [16, 10, 10, 38, 10, 10];
+  const line = values
+    .map((value, index) => value.padEnd(widths[index] ?? value.length))
+    .join('  ')
+    .trimEnd();
+  console.log(line);
+}
+
+printRow(['Scenario', 'Full doc', 'Event', 'Benchmark', 'Ops/s', 'MiB/s']);
+printRow(['--------', '--------', '-----', '---------', '-----', '-----']);
+
+for (const operationType of OPERATION_TYPES) {
+  for (const size of SIZE_TARGETS) {
+    const scenario = buildScenario(operationType, size.label, size.bytes);
+    const iterations = chooseIterations(scenario.eventBytes);
+    const results = BENCHMARKS.map((benchmark) => ({
+      label: benchmark.label,
+      ...runBenchmark(benchmark.label, benchmark.run, scenario.buffer, iterations)
+    }));
+
+    let isFirstRow = true;
+    for (const result of results) {
+      printRow([
+        isFirstRow ? scenario.label : '',
+        isFirstRow ? formatBytes(scenario.fullDocumentBytes) : '',
+        isFirstRow ? formatBytes(scenario.eventBytes) : '',
+        result.label,
+        formatNumber(result.opsPerSecond),
+        formatNumber(result.mibPerSecond)
+      ]);
+      isFirstRow = false;
+    }
+  }
+}
+
+function formatNumber(value: number): string {
+  return new Intl.NumberFormat('en-US', {
+    maximumFractionDigits: value >= 100 ? 0 : 1
+  }).format(value);
+}
+
+function formatBytes(bytes: number): string {
+  if (bytes < 1024) {
+    return `${bytes} B`;
+  }
+  if (bytes < 1024 * 1024) {
+    return `${(bytes / 1024).toFixed(bytes >= 10 * 1024 ? 0 : 1)} KB`;
+  }
+  return `${(bytes / (1024 * 1024)).toFixed(1)} MB`;
+}

modules/module-mongodb/src/replication/ChangeStream.ts

@@ -33,7 +33,7 @@ import {
   rawChangeStream
 } from './RawChangeStream.js';
 import { CHECKPOINTS_COLLECTION, timestampToDate } from './replication-utils.js';
-import { DefaultSourceRowConverter, SourceRowConverter } from './SourceRowConverter.js';
+import { DirectSourceRowConverter, SourceRowConverter } from './SourceRowConverter.js';
 
 export interface ChangeStreamOptions {
   connections: MongoManager;
@@ -118,7 +118,7 @@ export class ChangeStream {
     this.sync_rules = options.storage.getParsedSyncRules({
       defaultSchema: this.defaultDb.databaseName
     });
-    this.sourceRowConverter = new DefaultSourceRowConverter(this.sync_rules.compatibility);
+    this.sourceRowConverter = new DirectSourceRowConverter(this.sync_rules.compatibility);
 
     // The change stream aggregation command should timeout before the socket times out,
     // so we use 90% of the socket timeout value.