[FLINK-925] Extended for distinct operator and added test cases

This closes #59