fix: throw away excess data in order to avoid delivering duplicate data

src/table.ts

@@ -763,6 +763,19 @@
           callback();
           return;
         }
+        if (TableUtils.lessThanOrEqualTo(row.id, lastRowKey)) {
+          /*
+          Sometimes duplicate rows reach this point. To avoid delivering
+          duplicate rows to the user, rows are thrown away if they don't exceed
+          the last row key. We can expect each row to reach this point and rows
+          are delivered in order so if the last row key equals or exceeds the
+          row id then we know data for this row has already reached this point
+          and been delivered to the user. In this case we want to throw the row
+          away and we do not want to deliver this row to the user again.
+           */
+          callback();
+          return;
+        }
         lastRowKey = row.id;
         rowsRead++;
         callback(null, row);
@@ -817,7 +830,7 @@
         // Handling retries in this client. Specify the retry options to
         // make sure nothing is retried in retry-request.
         noResponseRetries: 0,
         shouldRetryFn: (_: any) => {
           return false;
         },
       };
@@ -975,7 +988,7 @@
             userStream.emit('error', error);
           }
         })
         .on('data', _ => {
           // Reset error count after a successful read so the backoff
           // time won't keep increasing when as stream had multiple errors
           numConsecutiveErrors = 0;
@@ -1598,7 +1611,7 @@
         // Handling retries in this client. Specify the retry options to
         // make sure nothing is retried in retry-request.
         noResponseRetries: 0,
         shouldRetryFn: (_: any) => {
           return false;
         },
       };

test/table.ts

@@ -934,8 +934,13 @@ describe('Bigtable/Table', () => {
       ];
 
       beforeEach(() => {
-        sinon.stub(table, 'row').callsFake(() => {
-          return {} as Row;
+        sinon.stub(table, 'row').callsFake((...args: unknown[]) => {
+          return {
+            id: args[0] as string,
+            table: table,
+            bigtable: table.bigtable,
+            data: {},
+          } as Row;
         });
         FakeChunkTransformer.prototype._transform = function (
           chunks: Array<{}>,