[FLINK-3463] implement calc translation

- remove FlinkFilter, FlinkProject and associated rules

- deactivate FilterReduceExpressionsRule and ProjectReduceExpressionsRule
  (covered by CalcReduceExpressions)

This closes apache#1696

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala

@@ -62,7 +62,7 @@ class DataSetFlatMap(
 
   override def translateToPlan(config: TableConfig,
       expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
-    val inputDataSet = input.asInstanceOf[DataSetRel].translateToPlan(config)
+    val inputDataSet = input.asInstanceOf[DataSetRel].translateToPlan(config, expectedType)
     val returnType = determineReturnType(
       getRowType,
       expectedType,

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala

@@ -76,9 +76,7 @@ object FlinkRuleSets {
 
     // simplify expressions rules
     ReduceExpressionsRule.CALC_INSTANCE,
-    ReduceExpressionsRule.FILTER_INSTANCE,
     ReduceExpressionsRule.JOIN_INSTANCE,
-    ReduceExpressionsRule.PROJECT_INSTANCE,
 
     // prune empty results rules
     PruneEmptyRules.AGGREGATE_INSTANCE,
@@ -99,9 +97,7 @@ object FlinkRuleSets {
     // translate to logical Flink nodes
     FlinkAggregateRule.INSTANCE,
     FlinkCalcRule.INSTANCE,
-    FlinkFilterRule.INSTANCE,
     FlinkJoinRule.INSTANCE,
-    FlinkProjectRule.INSTANCE,
     FlinkScanRule.INSTANCE,
     FlinkUnionRule.INSTANCE
   )
@@ -111,9 +107,7 @@ object FlinkRuleSets {
     // translate to DataSet nodes
     DataSetAggregateRule.INSTANCE,
     DataSetCalcRule.INSTANCE,
-    DataSetFilterRule.INSTANCE,
     DataSetJoinRule.INSTANCE,
-    DataSetProjectRule.INSTANCE,
     DataSetScanRule.INSTANCE,
     DataSetUnionRule.INSTANCE
   )

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala

@@ -23,6 +23,14 @@ import org.apache.calcite.rel.RelNode
 import org.apache.calcite.rel.convert.ConverterRule
 import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetFlatMap}
 import org.apache.flink.api.table.plan.nodes.logical.{FlinkCalc, FlinkConvention}
+import org.apache.flink.api.table.runtime.FlatMapRunner
+import org.apache.flink.api.table.TableConfig
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.table.codegen.CodeGenerator
+import org.apache.flink.api.common.functions.FlatMapFunction
+import scala.collection.JavaConversions._
+import org.apache.calcite.rex.RexLocalRef
+import org.apache.flink.api.table.codegen.GeneratedExpression
 
 class DataSetCalcRule
   extends ConverterRule(
@@ -37,13 +45,90 @@ class DataSetCalcRule
     val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
     val convInput: RelNode = RelOptRule.convert(calc.getInput, DataSetConvention.INSTANCE)
 
+    val calcFunc = (
+        config: TableConfig,
+        inputType: TypeInformation[Any],
+        returnType: TypeInformation[Any]) => {
+      val generator = new CodeGenerator(config, inputType)
+
+      val calcProgram = calc.getProgram
+      val condition = calcProgram.getCondition
+      val expandedExpressions = calcProgram.getProjectList.map(
+          expr => calcProgram.expandLocalRef(expr.asInstanceOf[RexLocalRef]))
+      val projection = generator.generateResultExpression(
+        returnType,
+        calc.getRowType.getFieldNames,
+        expandedExpressions)
+
+      val body = {
+        // only projection
+        if (condition == null) {
+          s"""
+            |${projection.code}
+            |${generator.collectorTerm}.collect(${projection.resultTerm});
+            |""".stripMargin
+        }
+        else {
+          val filterCondition = generator.generateExpression(
+              calcProgram.expandLocalRef(calcProgram.getCondition))
+          // only filter
+          if (projection == null) {
+            // conversion
+            if (inputType != returnType) {
+              val conversion = generator.generateConverterResultExpression(
+                  returnType,
+                  calc.getRowType.getFieldNames)
+
+                  s"""
+                    |${filterCondition.code}
+                    |if (${filterCondition.resultTerm}) {
+                    |  ${conversion.code}
+                    |  ${generator.collectorTerm}.collect(${conversion.resultTerm});
+                    |}
+                    |""".stripMargin
+            }
+            // no conversion
+            else {
+              s"""
+                |${filterCondition.code}
+                |if (${filterCondition.resultTerm}) {
+                |  ${generator.collectorTerm}.collect(${generator.input1Term});
+                |}
+                |""".stripMargin
+            }
+          }
+          // both filter and projection
+          else {
+            s"""
+              |${filterCondition.code}
+              |if (${filterCondition.resultTerm}) {
+              |  ${projection.code}
+              |  ${generator.collectorTerm}.collect(${projection.resultTerm});
+              |}
+              |""".stripMargin
+          }
+        }
+      }
+
+      val genFunction = generator.generateFunction(
+        description,
+        classOf[FlatMapFunction[Any, Any]],
+        body,
+        returnType)
+
+      new FlatMapRunner[Any, Any](
+        genFunction.name,
+        genFunction.code,
+        genFunction.returnType)
+    }
+
     new DataSetFlatMap(
       rel.getCluster,
       traitSet,
       convInput,
       rel.getRowType,
       calc.toString,
-      null)
+      calcFunc)
   }
 }