[FLINK-1245] [Java API] Introduce TypeHints for Java API operators

Also contains fixes by [email protected]
 - Make MissingTypeInfo optional in TypeExtractor (by default still throws exception)
 - Simplified deferred evaluation of type dependend code by making evaluations lazy
 - Add call location function names to MissingTypeInfo error messages.
 - Improvements on other error messages.

This closes apache#270

flink-clients/src/test/java/org/apache/flink/client/testjar/WordCount.java

@@ -27,8 +27,7 @@
 import org.apache.flink.api.java.ExecutionEnvironment;
 
 /**
- * Wordcount for placing at least something into the jar file.
- * 
+ * WordCount for placing at least something into the jar file.
  */
 public class WordCount {
 
@@ -98,7 +97,7 @@ public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
  // *************************************************************************
 
  private static boolean fileOutput = false;
- private static boolean verbose = false;
+
  private static String textPath;
  private static String outputPath;
 
@@ -111,7 +110,7 @@ private static boolean parseParameters(String[] args) {
  textPath = args[0];
  outputPath = args[1];
  } else if(args.length == 4 && (args[0].startsWith("-v") || args[0].startsWith("--verbose"))) { // cli line: program {optArg} {optVal} {textPath} {outputPath}
- verbose = Boolean.valueOf(args[1]);
+ Boolean.valueOf(args[1]); // parse verbosity flag
  textPath = args[2];
  outputPath = args[3];
  } else {

flink-core/src/main/java/org/apache/flink/api/common/operators/DualInputSemanticProperties.java

@@ -56,8 +56,7 @@ public class DualInputSemanticProperties extends SemanticProperties {
 
 
  public DualInputSemanticProperties() {
- super();
- this.init();
+ init();
  }
 
  /**
@@ -251,15 +250,24 @@ public FieldSet getReadFields2() {
  */
  @Override
  public void clearProperties() {
- this.init();
  super.clearProperties();
+ init();
  }
 
+ @Override
+ public boolean isEmpty() {
+ return super.isEmpty() &&
+ (forwardedFields1 == null || forwardedFields1.isEmpty()) &&
+ (forwardedFields2 == null || forwardedFields2.isEmpty()) &&
+ (readFields1 == null || readFields1.size() == 0) &&
+ (readFields2 == null || readFields2.size() == 0);
+ }
+
+
  private void init() {
  this.forwardedFields1 = new HashMap<Integer,FieldSet>();
  this.forwardedFields2 = new HashMap<Integer,FieldSet>();
  this.readFields1 = null;
  this.readFields2 = null;
  }
-
 }

flink-core/src/main/java/org/apache/flink/api/common/operators/SemanticProperties.java

@@ -30,11 +30,10 @@ public abstract class SemanticProperties implements Serializable {
 
  private static final long serialVersionUID = 1L;
 
- /**
- * Set of fields that are written in the destination record(s).
- */
+ /** Set of fields that are written in the destination record(s).*/
  private FieldSet writtenFields;
 
+
  /**
  * Adds, to the existing information, field(s) that are written in
  * the destination record(s).
@@ -71,10 +70,10 @@ public FieldSet getWrittenFields() {
  * Clears the object.
  */
  public void clearProperties() {
- this.init();
+ this.writtenFields = null;
  }
 
- private void init() {
- this.writtenFields = null;
+ public boolean isEmpty() {
+ return this.writtenFields == null || this.writtenFields.size() == 0;
  }
 }

flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputSemanticProperties.java

@@ -27,23 +27,18 @@
  * Container for the semantic properties associated to a single input operator.
  */
 public class SingleInputSemanticProperties extends SemanticProperties {
+
  private static final long serialVersionUID = 1L;
 
- /**
- * Mapping from fields in the source record(s) to fields in the destination
- * record(s). 
- */
+ /**Mapping from fields in the source record(s) to fields in the destination record(s). */
  private Map<Integer,FieldSet> forwardedFields;
 
- /**
- * Set of fields that are read in the source record(s).
- */
+ /** Set of fields that are read in the source record(s).*/
  private FieldSet readFields;
 
 
  public SingleInputSemanticProperties() {
- super();
- this.init();
+ init();
  }
 
  /**
@@ -140,8 +135,15 @@ public FieldSet getReadFields() {
  */
  @Override
  public void clearProperties() {
- this.init();
  super.clearProperties();
+ init();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return super.isEmpty() &&
+ (forwardedFields == null || forwardedFields.isEmpty()) &&
+ (readFields == null || readFields.size() == 0);
  }
 
  private void init() {
@@ -206,5 +208,10 @@ public void addWrittenFields(FieldSet writtenFields) {
  public void setWrittenFields(FieldSet writtenFields) {
  throw new UnsupportedOperationException();
  }
+
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
  }
 }