[FLINK-1237] Add support for custom partitioners

  - Functions: GroupReduce, Reduce, Aggregate on UnsortedGrouping, SortedGrouping (Java API & Scala API)
  - Manual partition on DataSet (Java API & Scala API)
  - Distinct operations provide semantic properties for preservation of distinctified fields
  - Tests for pushown (or not pushdown) of custom partitionings and forced rebalancing
  - Tests for GlobalProperties matching of partitionings
  - Caching of generated requested data properties for unary operators
Does it make sense to add another method ``getMaxNumPartititions()`` to check at plan construction time if the the DOP is high enough for the partitioner? Otherwise the program will fail at runtime, if I'm not mistaken.