apache/incubator-flink — pull request #85 by ktzoumas (2014-07-29 16:42:20+00:00)
Title: Java api functions to sams

This is FLINK-701. This is not good to merge yet, I am putting it out for comments and assistance. However, it should be part of release 0.6 as it breaks compatibility.

The patch changes the Java and Record APIs to work on top of SAM (Single Abstract Method) interfaces rather than abstract classes. The SAMs are named Fooable for an operation Foo (e.g., Mappable for Map, Reducible for Reduce), and they replace the former GenericFooer (e.g., GenericMapper) interfaces.

The original "rich functions" still exist and work as usual, as they implement hte aforementioned interfaces. They are called FooFunction (e.g., MapFunction, ReduceFunction), and contain the open(), close(), etc methods as well.

As part of the refactoring, Cross was changed to return exactly one value rather than taking a collector as input.

GenericCombiner is renamed to FlatCombinable (general naming rule: FlatFooable takes a Collector as parameter and does not return a value, Fooable returns exactly one value if both interfaces
exist). This PR does *not* add a Combinable and does *not* solve FLINK-848, these can be added later without breaking compatibility.

This PR does add an explicit FlatJoinable that is at the same level of inheritance as Joiable. The runtime works only on FlatJoinable objects, Joinables are shallowly transformed to FlatJoinables (see
GeneratedFlatJoinFunction).

Two consequences:

(1) FlatCross is removed from the Scala API, as it cannot be supported by the new Cross signature. This is an API design choice that cannot be rectified immediately.

(2) As a side-effect, this PR does add support for Java 8 lambdas in the filter and reduce operators. Lambdas in the other operators do not work yet, as the current TypeExctractor implementation needs to be adapted to extract the return types of lambdas (filter and reduce have both known return types). With that solved, FLINK-612 will probably be resolved as well.

Things that need to be fixed:

(1) Currently, several POM files require Java 8. This will be changed. The goal is to have some lambda tests that are executed only if Java 8 is present.

(2) If a lambda function is currently provided as an argument to a method other than filter or reduce, an error is displayed to the user. Detecting that the input is a lambda is currently done via string matching on the function name (see FunctionUtils.isLambdaFunction) which (a) is quite iffy, and (b) might not work with all JVMSs/break in several occasions.

If someone knows a good way to "detect if something is a lambda", that would be great! The internet was not too much help until now.

(3) Could I have an extra set of eyes on RegularPactTask, line 531?

(4) DeltaIterationTranslationTest currently does not pass, this is simple fix, I will do this asap.