This could be implemented by adapting the DBSP operator DistinctIncrementalTotal.
This could be implemented by adapting the DBSP operator DistinctIncrementalTotal.