feat: add stats on row duplication

Author: livetocode

README.md

@@ -327,8 +327,8 @@ You can resolve the conflict by keeping the first or last row of the duplicates:
 ```Typescript
 import { diff } from 'tabular-data-differ';
 const stats = await diff({
-    oldSource: './tests/a.csv',
-    newSource: './tests/b.csv',
+    oldSource: './tests/a2.csv',
+    newSource: './tests/b2.csv',
     keys: ['id'],
     duplicateKeyHandling: 'keepFirstRow', // or 'keepLastRow'
 }).to('console');
@@ -339,17 +339,34 @@ Or, if you need more control in the row selection, then you can provide your own
 ```Typescript
 import { diff } from 'tabular-data-differ';
 const stats = await diff({
-    oldSource: './tests/a.csv',
-    newSource: './tests/b.csv',
+    oldSource: './tests/a2.csv',
+    newSource: './tests/b2.csv',
     keys: ['id'],
     duplicateKeyHandling: (rows) => rows[0], // same as 'keepFirstRow'
     duplicateRowBufferSize: 2000,
-}).to('console');
+}).to('null');
 console.log(stats);
 ```
 
 Note that you can specify the size of the buffer if you know that it cannot exceed this quantity, otherwise you can enable the **duplicateRowBufferOverflow** option,
-which will remove the first entries when it exceeds the allocated capacity, to avoid a failure.
+which will remove the first entries when it exceeds the allocated capacity, to avoid any failure.
+
+Finally, you can inspect the source stats to check the duplication metrics:
+```Typescript
+import { diff } from 'tabular-data-differ';
+const ctx = await diff({
+    oldSource: './tests/a2.csv',
+    newSource: './tests/b2.csv',
+    keys: ['id'],
+    duplicateKeyHandling: 'keepFirstRow', // or 'keepLastRow'
+}).start();
+const stats = await ctx.to('null');
+console.log(stats);
+console.log(ctx.oldStats);
+console.log(ctx.newStats);
+
+```
+
 
 ### Order 2 CSV files and diff them on the console
 
@@ -632,6 +649,14 @@ Returns the current column names.
 
 Returns the currents stats.
 
+#### oldSourceStats
+
+Returns the stats accumulated while parsing the old source.
+
+#### newSourceStats
+
+Returns the stats accumulated while parsing the new source.
+
 #### to
 
 Initiates the comparison between the old and new sources and sends the diffs to the specified output.

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "tabular-data-differ",
-  "version": "1.1.3",
+  "version": "1.1.4",
   "description": "A very efficient library for diffing two sorted streams of tabular data, such as CSV files.",
   "keywords": [
     "table",

package.json

@@ -168,7 +168,8 @@ describe('differ', () => {
                 keys: ['ID'],
             })).rejects.toThrowError(new UniqueKeyViolationError(`Expected rows to be unique by "ID" in old source but received:
   previous=3,dave,44
-  current=3,dave bis,444`));
+  current=3,dave bis,444
+Note that you can resolve this conflict automatically using the duplicateKeyHandling option.`));
         });
         test('should detect primary key violation in new source', async () => {
             await expect(() => diffStrings({
@@ -189,7 +190,8 @@ describe('differ', () => {
                 keys: ['ID'],
             })).rejects.toThrowError(new UniqueKeyViolationError(`Expected rows to be unique by "ID" in new source but received:
   previous=3,dave,44
-  current=3,dave bis,444`));
+  current=3,dave bis,444
+Note that you can resolve this conflict automatically using the duplicateKeyHandling option.`));
         });
         test('should detect duplicate keys and return the first row', async () => {
             const writer = await diffStrings({
@@ -1849,7 +1851,8 @@ describe('differ', () => {
                 keys: ['id'],
                 duplicateKeyHandling: 'keepFirstRow',
             });
-            await differ.to({
+            const ctx = await differ.start();
+            await ctx.to({
                 destination: {
                     format: 'custom',
                     writer: output,
@@ -1878,6 +1881,28 @@ describe('differ', () => {
                 modified: 1,
                 same: 5
             });
+            expect(ctx.oldSourceStats).toEqual({
+                parsedRows: 12,
+                duplicateParsedRows: 3,
+                uniqueRows: 9,
+                uniqueRowsWithDuplicates: 3,
+                duplicationPercent: 25,
+                uniqueRowDuplicationPercent: 33.3333,
+                maxDuplicatesPerUniqueKey: 1,
+                minDuplicatesPerUniqueKey: 1,
+                averageDuplicatesPerUniqueKey: 1                
+            });
+            expect(ctx.newSourceStats).toEqual({
+                parsedRows: 10,
+                duplicateParsedRows: 3,
+                uniqueRows: 7,
+                uniqueRowsWithDuplicates: 2,
+                duplicationPercent: 30,
+                uniqueRowDuplicationPercent: 28.5714,
+                maxDuplicatesPerUniqueKey: 2,
+                minDuplicatesPerUniqueKey: 1,
+                averageDuplicatesPerUniqueKey: 1.5                 
+            });
         });
         test('should work with http streams (CSV)', async () => {
             const currentDir = process.cwd().replaceAll('\\', '/');

src/differ.test.ts

@@ -30,6 +30,7 @@ import {
     IterableFormatReaderOptions,
     IterableFormatReader,
     BufferedFormatReader,
+    roundDecimals,
 } from "./formats";
 
 export class UnorderedStreamsError extends Error {
@@ -171,6 +172,57 @@ export type DuplicateKeyHandler = (rows: Row[]) => Row;
 
 export type DuplicateKeyHandling = 'fail' |'keepFirstRow' | 'keepLastRow' | DuplicateKeyHandler;
 
+export class SourceStats {
+    parsedRows = 0;
+    duplicateParsedRows = 0;
+    uniqueRows = 0;
+    uniqueRowsWithDuplicates = 0;
+
+    duplicationPercent = 0;
+    uniqueRowDuplicationPercent = 0;
+    
+    maxDuplicatesPerUniqueKey = 0;
+    minDuplicatesPerUniqueKey = 0;
+    averageDuplicatesPerUniqueKey = 0;
+    
+    incParsedRows() {
+        this.parsedRows += 1;
+    }
+    
+    incDuplicateParsedRows() {
+        this.duplicateParsedRows += 1;
+    }
+    
+    incUniqueRows() {
+        this.uniqueRows += 1;
+    }
+    
+    incUniqueRowsWithDuplicates() {
+        this.uniqueRowsWithDuplicates += 1;
+    }
+
+    incDuplicates(value: number) {
+        this.maxDuplicatesPerUniqueKey = Math.max(this.maxDuplicatesPerUniqueKey, value);
+        if (this.minDuplicatesPerUniqueKey === 0) {
+            this.minDuplicatesPerUniqueKey = value;
+        } else {
+            this.minDuplicatesPerUniqueKey = Math.min(this.minDuplicatesPerUniqueKey, value);
+        }
+    }
+
+    calcStats() {
+        if (this.uniqueRowsWithDuplicates) {
+            this.averageDuplicatesPerUniqueKey = roundDecimals(this.duplicateParsedRows / this.uniqueRowsWithDuplicates, 4);
+        }
+        if (this.parsedRows) {
+            this.duplicationPercent = roundDecimals((this.duplicateParsedRows / this.parsedRows) * 100, 4);
+        }
+        if ( this.uniqueRows) {
+            this.uniqueRowDuplicationPercent = roundDecimals((this.uniqueRowsWithDuplicates / this.uniqueRows) * 100, 4);
+        }
+    }
+}
+
 /**
  * Options for configuring the Differ object that will traverse two input streams in parallel in order to compare their rows
  * and produce a change set.
@@ -378,6 +430,8 @@ export class DifferContext {
     private normalizeNewRow: RowNormalizer = row => row;
     private duplicateKeyHandling: DuplicateKeyHandling;
     private duplicateRowBufferSize: number;
+    private _oldSourceStats = new SourceStats();
+    private _newSourceStats = new SourceStats();
 
     constructor(private options: DifferOptions) {
         this.oldSource = new BufferedFormatReader(createSource(options.oldSource));
@@ -394,6 +448,8 @@ export class DifferContext {
     async [OpenSymbol](): Promise<void> {
         if (!this._isOpen) {
             this._isOpen = true;
+            this._oldSourceStats = new SourceStats();
+            this._newSourceStats = new SourceStats();
             await this.oldSource.open();
             await this.newSource.open();
             await this.extractHeaders();
@@ -436,6 +492,22 @@ export class DifferContext {
     get stats(): DiffStats {
         return this._stats;
     }
+
+    /**
+     * gets the stats accumulated while parsing the old source
+     * @returns the source stats
+     */
+    get oldSourceStats(): SourceStats {
+        return this._oldSourceStats;
+    }
+
+    /**
+     * gets the stats accumulated while parsing the new source
+     * @returns the source stats
+     */
+    get newSourceStats(): SourceStats {
+        return this._newSourceStats;
+    }
 
     /**
      * Iterates over the changes and sends them to the submitted output.
@@ -542,6 +614,8 @@ export class DifferContext {
                 previousPair = pair;
             }
         } finally {
+            this.oldSourceStats.calcStats();
+            this.newSourceStats.calcStats();
             this.close();
         }
     }
@@ -609,11 +683,13 @@ export class DifferContext {
         return result;
     }
 
-    async getNextRow(source: BufferedFormatReader): Promise<Row | undefined> {        
+    async getNextRow(source: BufferedFormatReader, stats: SourceStats): Promise<Row | undefined> {        
         const row = await source.readRow();
         if (!row) {
             return row;
         }
+        stats.incParsedRows();
+        stats.incUniqueRows();
         if (this.duplicateKeyHandling === 'fail') {
             // Note that it will be further processed in ensureRowsAreInAscendingOrder
             return row;
@@ -626,9 +702,14 @@ export class DifferContext {
         if (isDuplicate) {
             const duplicateRows: Row[] = [];
             duplicateRows.push(row);
+            stats.incUniqueRowsWithDuplicates();
+            let duplicateCount = 0;
             while(isDuplicate) {
                 const duplicateRow = await source.readRow();
                 if (duplicateRow) {
+                    duplicateCount += 1;
+                    stats.incParsedRows();
+                    stats.incDuplicateParsedRows();
                     if (this.duplicateKeyHandling !== 'keepFirstRow') {
                         // we don't need to accumulate duplicate rows when we just have to return the first row!
                         duplicateRows.push(duplicateRow);
@@ -649,6 +730,7 @@ export class DifferContext {
                 const nextRow = await source.peekRow();                        
                 isDuplicate = !!nextRow && this.comparer(this.keys, nextRow, row) === 0;
             }
+            stats.incDuplicates(duplicateCount);
             if (this.duplicateKeyHandling === 'keepFirstRow') {
                 return duplicateRows[0];
             }
@@ -661,11 +743,11 @@ export class DifferContext {
     }
 
     private getNextOldRow(): Promise<Row | undefined> {        
-        return this.getNextRow(this.oldSource);
+        return this.getNextRow(this.oldSource, this._oldSourceStats);
     }
 
     private getNextNewRow(): Promise<Row | undefined> {
-        return this.getNextRow(this.newSource);
+        return this.getNextRow(this.newSource, this._newSourceStats);
     }
 
     private async getNextPair():Promise<RowPair> {
@@ -693,7 +775,7 @@ export class DifferContext {
             const oldDelta = this.comparer(this.keys, previous, current);
             if (oldDelta === 0) {
                 const cols = this.keys.map(key => key.name);
-                throw new UniqueKeyViolationError(`Expected rows to be unique by "${cols}" in ${source} source but received:\n  previous=${previous}\n  current=${current}`);    
+                throw new UniqueKeyViolationError(`Expected rows to be unique by "${cols}" in ${source} source but received:\n  previous=${previous}\n  current=${current}\nNote that you can resolve this conflict automatically using the duplicateKeyHandling option.`);    
             }
             if (oldDelta > 0) {
                 const colOrder = this.keys.map(key => `${key.name} ${key.sortDirection ?? 'ASC'}`);