diff --git a/CHANGELOG.md b/CHANGELOG.md index ab9a508..4bee7c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## 2.0.0 (release candidate) +- Add `increment(key, incrementBy = 1)` for nonnegative uint32 weights, with atomic overflow rejection and collision-correct conservative updates. Supersedes the weighted-increment proposal in #1; credit to Ruslan Dzhumakaliev. + - Count keys such as `__proto__`, `constructor` and `toString` without interacting with Object.prototype. - Return independent top-k tuples so callers cannot modify internal sketch state. - Match serialized-size estimates to the actual power-of-two width. diff --git a/README.md b/README.md index e74c1ac..b0c7a78 100644 --- a/README.md +++ b/README.md @@ -243,3 +243,17 @@ Version 2 writes capacity-preserving CMS2 sketches by default. Old files remain TypeScript declarations cover the package root and existing class/helper deep imports. Node TypeScript projects need `@types/node`. Runtime compatibility starts at Node 6; development and the complete test suite use Node 22 or newer. CI runs the full suite on 22/24/26 and a separate Node 6 runtime smoke test. Before releasing: `npm ci`, `npm test`, then review the archive produced by `npm pack`. The test suite itself installs the archive into an independent consumer, checks CommonJS/ESM and compiles NodeNext/Node16 TypeScript fixtures. `npm publish` runs the tests through `prepublishOnly`; publishing still requires the maintainer's authenticated release action. The version bump in this branch is preparation, not evidence of a published release. + +### Weighted observations + +`counter.increment(key, incrementBy = 1)` accepts a nonnegative integer weight up to 4,294,967,295. Zero is a no-op after key validation. Negative/fractional/nonfinite values, other types and an update exceeding the uint32 counter limit throw before mutation. The implementation raises every selected sketch bucket below `minimum + incrementBy`, matching repeated unit conservative updates for that key without looping over the weight. + +```js +const views = streamcount.createViewsCounter(10); +views.increment('page-a', 250); +views.increment('page-a'); // 251 +``` + +Weighted observations can aggregate complete per-key counts. Replaying only each worker's top-k list loses omitted keys and is not a general merge of CountMinSketch states. Conservative sketches also cannot be assumed to equal a sketch of the concatenated stream by simply adding their cells. + +The weighted-increment API was proposed by Ruslan Dzhumakaliev in [PR #1](https://github.com/jhurliman/node-streamcount/pull/1). This implementation retains that use case while adding conservative-update correctness, validation, overflow handling and type coverage. diff --git a/index.d.ts b/index.d.ts index 2a13e8d..e43dc85 100644 --- a/index.d.ts +++ b/index.d.ts @@ -4,7 +4,8 @@ export interface DeserializeOptions { maxEntries?: number; } /** Constructors return structural counter objects, not instanceof-compatible instances. */ export class CountMinSketch { constructor(maxEntries: number, epsilon: number, delta: number); - increment(key: string): void; + /** Add a nonnegative uint32 integer weight (default 1); zero is a no-op. */ + increment(key: string, incrementBy?: number): void; getTopK(): Array<[count: number, key: string]>; serialize(options?: SerializeOptions): Buffer; static deserialize(buffer: Buffer, start?: number, length?: number, options?: DeserializeOptions): CountMinSketch; diff --git a/lib/countMinSketch.js b/lib/countMinSketch.js index 7ae2e88..9ccd9e5 100644 --- a/lib/countMinSketch.js +++ b/lib/countMinSketch.js @@ -23,20 +23,26 @@ function create(maxEntries, lgWidth, counts, hashes, heap) { heap.forEach(function(entry) { map[entry[1]] = entry; }); var heapq = new MinHeap(heap, function(a, b) { return a[0] - b[0]; }); - function increment(key) { + function increment(key, incrementBy) { valid.key(key); + incrementBy = incrementBy === undefined ? 1 : incrementBy; + valid.integer(incrementBy, 0, MAX_INT, 'incrementBy'); + if (incrementBy === 0) return; var ix = hashing.fnv1a(key), est = MAX_INT; var i, j; for (i = 0; i < hashes.length; i++) { j = bucket(lgWidth, hashes[i], ix); est = Math.min(est, counts[i][j]); } - if (est === MAX_INT) throw new RangeError('counter would overflow uint32'); + if (incrementBy > MAX_INT - est) throw new RangeError('counter would overflow uint32'); + var target = est + incrementBy; for (i = 0; i < hashes.length; i++) { j = bucket(lgWidth, hashes[i], ix); - if (counts[i][j] === est) counts[i][j] = est + 1; + // Equivalent to repeated unit conservative updates: raise every + // selected counter below the new minimum, preserving higher collisions. + if (counts[i][j] < target) counts[i][j] = target; } - est++; + est = target; var probe = map[key]; if (probe !== undefined) { probe[0] = est; diff --git a/test/package-test.js b/test/package-test.js index ddd6a3e..3a3783c 100644 --- a/test/package-test.js +++ b/test/package-test.js @@ -23,7 +23,9 @@ import HLL = require('streamcount/lib/hyperLogLog'); import Heap = require('streamcount/lib/minHeap'); import Random = require('streamcount/lib/prng'); const views: api.CountMinSketch = new CMS(10, .1, .1); -views.increment('key'); +views.increment('key'); views.increment('key', 5); +// @ts-expect-error weights must be numbers +views.increment('key', '5'); const rows: Array<[number,string]> = views.getTopK(); const restored = CMS.deserialize(views.serialize({legacy:true}), undefined, undefined, {maxEntries:10}); const uniques: api.HyperLogLog = new HLL(.1); uniques.merge(api.createUniquesCounter()); diff --git a/test/runtime-smoke.js b/test/runtime-smoke.js index 624204b..07c8d03 100644 --- a/test/runtime-smoke.js +++ b/test/runtime-smoke.js @@ -8,3 +8,7 @@ assert.strictEqual(restored.getTopK().length, 2); var hll = api.createUniquesCounter(); hll.add('key'); assert.ok(api.HyperLogLog.deserialize(hll.serialize()).count() > 0); assert.throws(function() { api.createViewsCounter(0); }); + +var weighted = require('..').createViewsCounter(3); +weighted.increment('weighted', 7); +if (weighted.getTopK()[0][0] !== 7) throw new Error('weighted increment'); diff --git a/test/weighted-test.js b/test/weighted-test.js new file mode 100644 index 0000000..935314c --- /dev/null +++ b/test/weighted-test.js @@ -0,0 +1,62 @@ +'use strict'; +const {test}=require('node:test'); +const assert=require('node:assert/strict'); +const CMS=require('../lib/countMinSketch'); +function sketch(capacity=10){return new CMS(capacity,.5,.2);} +function counters(buffer){const depth=buffer.readUInt32LE(12),width=buffer.readUInt32LE(16);return buffer.subarray(20,20+depth*width*4);} +function seededRows(first,second){ + const bytes=sketch().serialize(),width=bytes.readUInt32LE(16); + for(let row=0;row<2;row++)for(let col=0;col{ + const cms=sketch();const empty=cms.serialize();cms.increment('absent',0);assert.deepEqual(cms.serialize(),empty); + cms.increment('a');cms.increment('a',undefined);cms.increment('a',5); + assert.deepEqual(cms.getTopK(),[[7,'a']]);const before=cms.serialize();cms.increment('a',0);assert.deepEqual(cms.serialize(),before); +}); +test('weights must be nonnegative uint32 integers and invalid updates are atomic',()=>{ + const cms=sketch();cms.increment('a',3);const before=cms.serialize(); + for(const weight of [-1,.5,NaN,Infinity,-Infinity,4294967296,Number.MAX_SAFE_INTEGER,null,'2',true,{},[]]){ + assert.throws(()=>cms.increment('a',weight),RangeError);assert.deepEqual(cms.serialize(),before); + } + assert.throws(()=>cms.increment(123,0),TypeError); +}); +test('weighted conservative updates raise intermediate collision buckets',()=>{ + const cms=seededRows(5,7);cms.increment('a',10); + assert.deepEqual(cms.getTopK(),[[15,'a']]); + assert.deepEqual(CMS.deserialize(cms.serialize()).getTopK(),[[15,'a']]); + const unit=seededRows(5,7);for(let i=0;i<10;i++)unit.increment('a'); + assert.deepEqual(cms.serialize(),unit.serialize()); +}); +test('weighted updates preserve collision counters above the target',()=>{ + const cms=seededRows(5,100);cms.increment('a',10); + const width=cms.serialize().readUInt32LE(16),row=counters(cms.serialize()).subarray(width*4); + for(let i=0;i{ + const cms=sketch();cms.increment('max',4294967295);const before=cms.serialize(); + cms.increment('max',0);assert.deepEqual(cms.serialize(),before); + assert.throws(()=>cms.increment('max'),RangeError);assert.deepEqual(cms.serialize(),before); + const near=seededRows(4294967290,4294967292),snapshot=near.serialize(); + assert.throws(()=>near.increment('a',6),RangeError);assert.deepEqual(near.serialize(),snapshot); + near.increment('a',5);assert.deepEqual(CMS.deserialize(near.serialize()).getTopK(),[[4294967295,'a']]); +}); +test('collision-heavy weighted streams match repeated unit updates',()=>{ + const bulk=sketch(3),units=sketch(3); + let seed=123; + for(let i=0;i<400;i++){ + seed=(Math.imul(seed,1664525)+1013904223)>>>0; + const key=['__proto__','constructor','λ','a','b','c','d'][seed%7],weight=(seed>>>8)%19; + bulk.increment(key,weight);for(let j=0;ja[1].localeCompare(b[1]);assert.deepEqual(bulk.getTopK().sort(order),units.getTopK().sort(order)); + } +}); +test('weighted counts survive legacy and CMS2 serialization and further updates',()=>{ + const cms=sketch();cms.increment('weighted',20); + for(const options of [{},{legacy:true}]){ + const restored=CMS.deserialize(cms.serialize(options),undefined,undefined,options.legacy?{maxEntries:10}:undefined); + restored.increment('weighted',5);assert.deepEqual(restored.getTopK(),[[25,'weighted']]); + } +});