[FLINK-11153][dataset] Remove UdfAnalyzer

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

@@ -32,8 +32,11 @@
  *  - Warnings if a Function returns null
  *  - Warnings if a tuple access uses a wrong index
  *  - Information about the number of object creations (for manual optimization)
+ *
+ * @deprecated The code analysis code has been removed and this enum has no effect.
  */
 @PublicEvolving
+@Deprecated
 public enum CodeAnalysisMode {
 
 	/**

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

@@ -56,10 +56,6 @@
  *         handling <i>generic types</i> and <i>POJOs</i>. This is usually only needed
  *         when the functions return not only the types declared in their signature, but
  *         also subclasses of those types.</li>
- *     <li>The {@link CodeAnalysisMode} of the program: Enable hinting/optimizing or disable
- *         the "static code analyzer". The static code analyzer pre-interprets user-defined functions in order to
- *         get implementation insights for program improvements that can be printed to the log or
- *         automatically applied.</li>
  * </ul>
  */
 @Public
@@ -699,23 +695,18 @@ public boolean isObjectReuseEnabled() {
 	}
 
 	/**
-	 * Sets the {@link CodeAnalysisMode} of the program. Specifies to which extent user-defined
-	 * functions are analyzed in order to give the Flink optimizer an insight of UDF internals
-	 * and inform the user about common implementation mistakes. The static code analyzer pre-interprets
-	 * user-defined functions in order to get implementation insights for program improvements
-	 * that can be printed to the log, automatically applied, or disabled.
-	 * 
-	 * @param codeAnalysisMode see {@link CodeAnalysisMode}
+	 * @deprecated The code analysis code has been removed and this method has no effect.
 	 */
 	@PublicEvolving
+	@Deprecated
 	public void setCodeAnalysisMode(CodeAnalysisMode codeAnalysisMode) {
-		this.codeAnalysisMode = codeAnalysisMode;
 	}
 
 	/**
-	 * Returns the {@link CodeAnalysisMode} of the program.
+	 * @deprecated The code analysis code has been removed and this method does not return anything interesting.
 	 */
 	@PublicEvolving
+	@Deprecated
 	public CodeAnalysisMode getCodeAnalysisMode() {
 		return codeAnalysisMode;
 	}

flink-java/src/main/java/org/apache/flink/api/java/Utils.java

@@ -36,8 +36,6 @@
 import java.lang.reflect.Modifier;
 import java.util.Random;
 
-import static org.apache.flink.api.java.functions.FunctionAnnotation.SkipCodeAnalysis;
-
 /**
  * Utility class that contains helper methods to work with Java APIs.
  */
@@ -71,7 +69,6 @@ public static String getCallLocationName(int depth) {
 	 *
 	 * @param <T> Type of elements to count.
 	 */
-	@SkipCodeAnalysis
 	public static class CountHelper<T> extends RichOutputFormat<T> {
 
 		private static final long serialVersionUID = 1L;
@@ -108,7 +105,6 @@ public void close() {
 	 *
 	 * @param <T> Type of elements to count.
 	 */
-	@SkipCodeAnalysis
 	public static class CollectHelper<T> extends RichOutputFormat<T> {
 
 		private static final long serialVersionUID = 1L;
@@ -220,7 +216,6 @@ public String toString() {
 	 * {@link RichOutputFormat} for {@link ChecksumHashCode}.
 	 * @param <T>
 	 */
-	@SkipCodeAnalysis
 	public static class ChecksumHashCodeHelper<T> extends RichOutputFormat<T> {
 
 		private static final long serialVersionUID = 1L;

flink-java/src/main/java/org/apache/flink/api/java/functions/FunctionAnnotation.java

@@ -384,10 +384,12 @@ public class FunctionAnnotation {
 	 * printed to the log as hints, automatically applied, or disabled (see
 	 * {@link org.apache.flink.api.common.ExecutionConfig}).
 	 *
+	 * @deprecated The code analysis code has been removed and this annotation has no effect.
 	 */
 	@Target(ElementType.TYPE)
 	@Retention(RetentionPolicy.RUNTIME)
 	@PublicEvolving
+	@Deprecated
 	public @interface SkipCodeAnalysis {
 	}
 

flink-java/src/main/java/org/apache/flink/api/java/operators/CoGroupOperator.java

@@ -121,8 +121,6 @@ public CoGroupOperator(DataSet<I1> input1, DataSet<I2> input2, Keys<I1> keys1, K
 
 		this.keys1 = keys1;
 		this.keys2 = keys2;
-
-		UdfOperatorUtils.analyzeDualInputUdf(this, CoGroupFunction.class, defaultName, function, keys1, keys2);
 	}
 
 	@Override

flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java

@@ -92,8 +92,6 @@ public CrossOperator(DataSet<I1> input1, DataSet<I2> input2,
 		this.function = function;
 		this.defaultName = defaultName;
 		this.hint = hint;
-
-		UdfOperatorUtils.analyzeDualInputUdf(this, CrossFunction.class, defaultName, function, null, null);
 	}
 
 	@Override

flink-java/src/main/java/org/apache/flink/api/java/operators/FilterOperator.java

@@ -43,8 +43,6 @@ public FilterOperator(DataSet<T> input, FilterFunction<T> function, String defau
 
 		this.function = function;
 		this.defaultName = defaultName;
-
-		UdfOperatorUtils.analyzeSingleInputUdf(this, FilterFunction.class, defaultName, function, null);
 	}
 
 	@Override

flink-java/src/main/java/org/apache/flink/api/java/operators/FlatMapOperator.java

@@ -45,8 +45,6 @@ public FlatMapOperator(DataSet<IN> input, TypeInformation<OUT> resultType, FlatM
 
 		this.function = function;
 		this.defaultName = defaultName;
-
-		UdfOperatorUtils.analyzeSingleInputUdf(this, FlatMapFunction.class, defaultName, function, null);
 	}
 
 	@Override

flink-java/src/main/java/org/apache/flink/api/java/operators/GroupReduceOperator.java

@@ -98,8 +98,6 @@ public GroupReduceOperator(Grouping<IN> input, TypeInformation<OUT> resultType,
 		this.defaultName = defaultName;
 
 		this.combinable = checkCombinability();
-
-		UdfOperatorUtils.analyzeSingleInputUdf(this, GroupReduceFunction.class, defaultName, function, grouper.keys);
 	}
 
 	private boolean checkCombinability() {

flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java

@@ -254,8 +254,6 @@ public EquiJoin(DataSet<I1> input1, DataSet<I2> input2,
 
 			this.function = function;
 			this.joinLocationName = joinLocationName;
-
-			UdfOperatorUtils.analyzeDualInputUdf(this, FlatJoinFunction.class, joinLocationName, function, keys1, keys2);
 		}
 
 		public EquiJoin(DataSet<I1> input1, DataSet<I2> input2,
@@ -270,8 +268,6 @@ public EquiJoin(DataSet<I1> input1, DataSet<I2> input2,
 			}
 
 			this.function = generatedFunction;
-
-			UdfOperatorUtils.analyzeDualInputUdf(this, JoinFunction.class, joinLocationName, function, keys1, keys2);
 		}
 
 		@Override

flink-java/src/main/java/org/apache/flink/api/java/operators/MapOperator.java

@@ -47,8 +47,6 @@ public MapOperator(DataSet<IN> input, TypeInformation<OUT> resultType, MapFuncti
 
 		this.defaultName = defaultName;
 		this.function = function;
-
-		UdfOperatorUtils.analyzeSingleInputUdf(this, MapFunction.class, defaultName, function, null);
 	}
 
 	@Override

flink-java/src/main/java/org/apache/flink/api/java/operators/ReduceOperator.java

@@ -77,8 +77,6 @@ public ReduceOperator(Grouping<IN> input, ReduceFunction<IN> function, String de
 		this.grouper = input;
 		this.defaultName = defaultName;
 		this.hint = CombineHint.OPTIMIZER_CHOOSES;
-
-		UdfOperatorUtils.analyzeSingleInputUdf(this, ReduceFunction.class, defaultName, function, grouper.keys);
 	}
 
 	@Override