You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2016/03/11 00:08:24 UTC
[1/3] flink git commit: [FLINK-3596] DataSet RelNode refactoring
Repository: flink
Updated Branches:
refs/heads/tableOnCalcite e34e43954 -> d720b002a
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala
index 1b876da..11857df 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala
@@ -27,7 +27,8 @@ import org.apache.calcite.sql.`type`.{SqlTypeFactoryImpl, SqlTypeName}
import org.apache.calcite.sql.fun._
import org.apache.flink.api.common.functions.{GroupReduceFunction, MapFunction}
import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.table.plan.PlanGenException
+import org.apache.flink.api.table.plan.{TypeConverter, PlanGenException}
+import org.apache.flink.api.table.plan.TypeConverter._
import org.apache.flink.api.table.typeinfo.RowTypeInfo
import org.apache.flink.api.table.{Row, TableConfig}
@@ -64,7 +65,8 @@ object AggregateUtil {
*/
def createOperatorFunctionsForAggregates(namedAggregates: Seq[CalcitePair[AggregateCall, String]],
inputType: RelDataType, outputType: RelDataType,
- groupings: Array[Int]): AggregateResult = {
+ groupings: Array[Int],
+ config: TableConfig): AggregateResult = {
val aggregateFunctionsAndFieldIndexes =
transformToAggregateFunctions(namedAggregates.map(_.getKey), inputType, groupings.length)
@@ -72,20 +74,19 @@ object AggregateUtil {
val aggFieldIndexes = aggregateFunctionsAndFieldIndexes._1
val aggregates = aggregateFunctionsAndFieldIndexes._2
- val mapFunction = (
- config: TableConfig,
- inputType: TypeInformation[Any],
- returnType: TypeInformation[Any]) => {
-
- val aggregateMapFunction = new AggregateMapFunction[Row, Row](
- aggregates, aggFieldIndexes, groupings, returnType.asInstanceOf[RowTypeInfo])
-
- aggregateMapFunction.asInstanceOf[MapFunction[Any, Any]]
- }
-
val bufferDataType: RelRecordType =
createAggregateBufferDataType(groupings, aggregates, inputType)
+ val mapReturnType = determineReturnType(
+ bufferDataType,
+ Some(TypeConverter.DEFAULT_ROW_TYPE),
+ config.getNullCheck,
+ config.getEfficientTypeUsage)
+
+ val mapFunction = new AggregateMapFunction[Row, Row](
+ aggregates, aggFieldIndexes, groupings,
+ mapReturnType.asInstanceOf[RowTypeInfo]).asInstanceOf[MapFunction[Any, Any]]
+
// the mapping relation between field index of intermediate aggregate Row and output Row.
val groupingOffsetMapping = getGroupKeysMapping(inputType, outputType, groupings)
@@ -105,16 +106,15 @@ object AggregateUtil {
val reduceGroupFunction =
if (allPartialAggregate) {
- (config: TableConfig, inputType: TypeInformation[Row], returnType: TypeInformation[Row]) =>
- new AggregateReduceCombineFunction(aggregates, groupingOffsetMapping,
- aggOffsetMapping, intermediateRowArity)
- } else {
- (config: TableConfig, inputType: TypeInformation[Row], returnType: TypeInformation[Row]) =>
- new AggregateReduceGroupFunction(aggregates, groupingOffsetMapping,
- aggOffsetMapping, intermediateRowArity)
+ new AggregateReduceCombineFunction(aggregates, groupingOffsetMapping,
+ aggOffsetMapping, intermediateRowArity)
+ }
+ else {
+ new AggregateReduceGroupFunction(aggregates, groupingOffsetMapping,
+ aggOffsetMapping, intermediateRowArity)
}
- new AggregateResult(mapFunction, reduceGroupFunction, bufferDataType)
+ new AggregateResult(mapFunction, reduceGroupFunction)
}
private def transformToAggregateFunctions(
@@ -318,9 +318,6 @@ object AggregateUtil {
}
case class AggregateResult(
- val mapFunc: (TableConfig, TypeInformation[Any], TypeInformation[Any]) =>
- MapFunction[Any, Any],
- val reduceGroupFunc: (TableConfig, TypeInformation[Row], TypeInformation[Row]) =>
- GroupReduceFunction[Row, Row],
- val intermediateDataType: RelDataType) {
+ val mapFunc: MapFunction[Any, Any],
+ val reduceGroupFunc: GroupReduceFunction[Row, Row]) {
}
[3/3] flink git commit: [FLINK-3596] Finalized DataSet RelNode
refactoring
Posted by fh...@apache.org.
[FLINK-3596] Finalized DataSet RelNode refactoring
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d720b002
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d720b002
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d720b002
Branch: refs/heads/tableOnCalcite
Commit: d720b002a77d4ac3a2e41bfaeaf6ad87c2e415c0
Parents: 22621e0
Author: Fabian Hueske <fh...@apache.org>
Authored: Thu Mar 10 23:30:06 2016 +0100
Committer: Fabian Hueske <fh...@apache.org>
Committed: Thu Mar 10 23:30:43 2016 +0100
----------------------------------------------------------------------
.../plan/nodes/dataset/DataSetAggregate.scala | 7 +-
.../table/plan/nodes/dataset/DataSetUnion.scala | 2 +-
.../api/table/plan/rules/FlinkRuleSets.scala | 12 +-
.../rules/dataSet/DataSetAggregateRule.scala | 55 +++++++
.../plan/rules/dataSet/DataSetCalcRule.scala | 53 ++++++
.../plan/rules/dataSet/DataSetJoinRule.scala | 115 +++++++++++++
.../plan/rules/dataSet/DataSetScanRule.scala | 50 ++++++
.../plan/rules/dataSet/DataSetUnionRule.scala | 54 ++++++
.../dataSet/FlinkJoinUnionTransposeRule.scala | 110 +++++++++++++
.../plan/rules/logical/FlinkAggregateRule.scala | 55 -------
.../plan/rules/logical/FlinkCalcRule.scala | 53 ------
.../plan/rules/logical/FlinkJoinRule.scala | 115 -------------
.../logical/FlinkJoinUnionTransposeRule.scala | 110 -------------
.../plan/rules/logical/FlinkScanRule.scala | 52 ------
.../plan/rules/logical/FlinkUnionRule.scala | 54 ------
.../table/runtime/aggregate/AggregateUtil.scala | 11 +-
.../test/GroupedAggregationsITCase.scala.orig | 164 -------------------
17 files changed, 450 insertions(+), 622 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
index 3856c5f..9a9bf99 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
@@ -24,7 +24,6 @@ import org.apache.calcite.rel.core.AggregateCall
import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.plan.TypeConverter._
import org.apache.flink.api.table.plan.{PlanGenException, TypeConverter}
import org.apache.flink.api.table.runtime.aggregate.AggregateUtil
import org.apache.flink.api.table.runtime.aggregate.AggregateUtil.CalcitePair
@@ -76,7 +75,7 @@ class DataSetAggregate(
case _ => // ok
}
- val groupingKeys = (0 until grouping.length).toArray
+ val groupingKeys = grouping.indices.toArray
// add grouping fields, position keys in the input, and input type
val aggregateResult = AggregateUtil.createOperatorFunctionsForAggregates(namedAggregates,
inputType, rowType, grouping, config)
@@ -93,8 +92,8 @@ class DataSetAggregate(
.toArray
val rowTypeInfo = new RowTypeInfo(fieldTypes)
- val mappedInput = inputDS.map(aggregateResult.mapFunc)
- val groupReduceFunction = aggregateResult.reduceGroupFunc
+ val mappedInput = inputDS.map(aggregateResult._1)
+ val groupReduceFunction = aggregateResult._2
if (groupingKeys.length > 0) {
mappedInput.asInstanceOf[DataSet[Row]]
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
index 462c4a5..a52a65e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
@@ -23,7 +23,7 @@ import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.{RelWriter, BiRel, RelNode}
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.{TableConfig, Row}
+import org.apache.flink.api.table.TableConfig
/**
* Flink RelNode which matches along with UnionOperator.
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
index bd128b2..d8cc2b3 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
@@ -20,7 +20,7 @@ package org.apache.flink.api.table.plan.rules
import org.apache.calcite.rel.rules._
import org.apache.calcite.tools.{RuleSets, RuleSet}
-import org.apache.flink.api.table.plan.rules.logical._
+import org.apache.flink.api.table.plan.rules.dataSet._
object FlinkRuleSets {
@@ -94,11 +94,11 @@ object FlinkRuleSets {
CalcMergeRule.INSTANCE,
// translate to logical Flink nodes
- FlinkAggregateRule.INSTANCE,
- FlinkCalcRule.INSTANCE,
- FlinkJoinRule.INSTANCE,
- FlinkScanRule.INSTANCE,
- FlinkUnionRule.INSTANCE
+ DataSetAggregateRule.INSTANCE,
+ DataSetCalcRule.INSTANCE,
+ DataSetJoinRule.INSTANCE,
+ DataSetScanRule.INSTANCE,
+ DataSetUnionRule.INSTANCE
)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetAggregateRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetAggregateRule.scala
new file mode 100644
index 0000000..40afd4a
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetAggregateRule.scala
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.rules.dataSet
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalAggregate
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetAggregate, DataSetConvention}
+import scala.collection.JavaConversions._
+
+class DataSetAggregateRule
+ extends ConverterRule(
+ classOf[LogicalAggregate],
+ Convention.NONE,
+ DataSetConvention.INSTANCE,
+ "FlinkAggregateRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val agg: LogicalAggregate = rel.asInstanceOf[LogicalAggregate]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(agg.getInput, DataSetConvention.INSTANCE)
+
+ new DataSetAggregate(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ agg.getNamedAggCalls,
+ rel.getRowType,
+ agg.getInput.getRowType,
+ agg.toString,
+ agg.getGroupSet.toArray)
+ }
+ }
+
+object DataSetAggregateRule {
+ val INSTANCE: RelOptRule = new DataSetAggregateRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetCalcRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetCalcRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetCalcRule.scala
new file mode 100644
index 0000000..3610819
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetCalcRule.scala
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.rules.dataSet
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalCalc
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetCalc, DataSetConvention}
+
+class DataSetCalcRule
+ extends ConverterRule(
+ classOf[LogicalCalc],
+ Convention.NONE,
+ DataSetConvention.INSTANCE,
+ "FlinkCalcRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val calc: LogicalCalc = rel.asInstanceOf[LogicalCalc]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(calc.getInput, DataSetConvention.INSTANCE)
+
+ new DataSetCalc(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ rel.getRowType,
+ calc.getProgram,
+ calc.toString,
+ description)
+ }
+ }
+
+object DataSetCalcRule {
+ val INSTANCE: RelOptRule = new DataSetCalcRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetJoinRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetJoinRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetJoinRule.scala
new file mode 100644
index 0000000..06f7d51
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetJoinRule.scala
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.rules.dataSet
+
+import org.apache.calcite.plan.{RelOptRuleCall, Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalJoin
+import org.apache.calcite.rex.{RexInputRef, RexCall}
+import org.apache.calcite.sql.fun.SqlStdOperatorTable
+import org.apache.flink.api.java.operators.join.JoinType
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetJoin, DataSetConvention}
+import scala.collection.JavaConverters._
+import scala.collection.JavaConversions._
+
+class DataSetJoinRule
+ extends ConverterRule(
+ classOf[LogicalJoin],
+ Convention.NONE,
+ DataSetConvention.INSTANCE,
+ "FlinkJoinRule")
+ {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+
+ val join = call.rel(0).asInstanceOf[LogicalJoin]
+ val children = join.getInputs
+ val rexBuilder = call.builder().getRexBuilder
+
+ val joinInfo = join.analyzeCondition()
+ val joinCondition = join.getCondition
+ val equiCondition =
+ joinInfo.getEquiCondition(children.get(0), children.get(1), rexBuilder)
+
+ // joins require at least one equi-condition
+ if (equiCondition.isAlwaysTrue) {
+ false
+ }
+ else {
+ // check that all equality predicates refer to field refs only (not computed expressions)
+ // Note: Calcite treats equality predicates on expressions as non-equi predicates
+ joinCondition match {
+
+ // conjunction of join predicates
+ case c: RexCall if c.getOperator.equals(SqlStdOperatorTable.AND) =>
+
+ c.getOperands.asScala
+ // look at equality predicates only
+ .filter { o =>
+ o.isInstanceOf[RexCall] &&
+ o.asInstanceOf[RexCall].getOperator.equals(SqlStdOperatorTable.EQUALS)
+ }
+ // check that both children are field references
+ .map { o =>
+ o.asInstanceOf[RexCall].getOperands.get(0).isInstanceOf[RexInputRef] &&
+ o.asInstanceOf[RexCall].getOperands.get(1).isInstanceOf[RexInputRef]
+ }
+ // any equality predicate that does not refer to a field reference?
+ .reduce( (a, b) => a && b)
+
+ // single equi-join predicate
+ case c: RexCall if c.getOperator.equals(SqlStdOperatorTable.EQUALS) =>
+ c.getOperands.get(0).isInstanceOf[RexInputRef] &&
+ c.getOperands.get(1).isInstanceOf[RexInputRef]
+ case _ =>
+ false
+ }
+ }
+
+ }
+
+ def convert(rel: RelNode): RelNode = {
+
+ val join: LogicalJoin = rel.asInstanceOf[LogicalJoin]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convLeft: RelNode = RelOptRule.convert(join.getInput(0), DataSetConvention.INSTANCE)
+ val convRight: RelNode = RelOptRule.convert(join.getInput(1), DataSetConvention.INSTANCE)
+ val joinInfo = join.analyzeCondition
+
+ new DataSetJoin(
+ rel.getCluster,
+ traitSet,
+ convLeft,
+ convRight,
+ rel.getRowType,
+ join.toString,
+ join.getCondition,
+ join.getRowType,
+ joinInfo,
+ joinInfo.pairs.toList,
+ JoinType.INNER,
+ null,
+ description)
+ }
+ }
+
+object DataSetJoinRule {
+ val INSTANCE: RelOptRule = new DataSetJoinRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetScanRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetScanRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetScanRule.scala
new file mode 100644
index 0000000..2865d9f
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetScanRule.scala
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.rules.dataSet
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.core.TableScan
+import org.apache.calcite.rel.logical.LogicalTableScan
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetSource}
+
+class DataSetScanRule
+ extends ConverterRule(
+ classOf[LogicalTableScan],
+ Convention.NONE,
+ DataSetConvention.INSTANCE,
+ "FlinkScanRule")
+ {
+ def convert(rel: RelNode): RelNode = {
+ val scan: TableScan = rel.asInstanceOf[TableScan]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+
+ new DataSetSource(
+ rel.getCluster,
+ traitSet,
+ scan.getTable,
+ rel.getRowType
+ )
+ }
+ }
+
+object DataSetScanRule {
+ val INSTANCE: RelOptRule = new DataSetScanRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetUnionRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetUnionRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetUnionRule.scala
new file mode 100644
index 0000000..6ab64c6
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetUnionRule.scala
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.rules.dataSet
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalUnion
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetUnion}
+
+class DataSetUnionRule
+ extends ConverterRule(
+ classOf[LogicalUnion],
+ Convention.NONE,
+ DataSetConvention.INSTANCE,
+ "FlinkUnionRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+
+ val union: LogicalUnion = rel.asInstanceOf[LogicalUnion]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convLeft: RelNode = RelOptRule.convert(union.getInput(0), DataSetConvention.INSTANCE)
+ val convRight: RelNode = RelOptRule.convert(union.getInput(1), DataSetConvention.INSTANCE)
+
+ new DataSetUnion(
+ rel.getCluster,
+ traitSet,
+ convLeft,
+ convRight,
+ rel.getRowType,
+ union.toString)
+ }
+ }
+
+object DataSetUnionRule {
+ val INSTANCE: RelOptRule = new DataSetUnionRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/FlinkJoinUnionTransposeRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/FlinkJoinUnionTransposeRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/FlinkJoinUnionTransposeRule.scala
new file mode 100644
index 0000000..32b53e2
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/FlinkJoinUnionTransposeRule.scala
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.rules.dataSet
+
+import org.apache.calcite.plan.RelOptRule.{any, operand, convert => convertTrait}
+import org.apache.calcite.plan.RelOptRule
+import org.apache.calcite.plan.RelOptRuleOperand
+import org.apache.calcite.plan.RelOptRuleCall
+import org.apache.calcite.rel.RelNode
+import java.util.ArrayList
+import scala.collection.JavaConversions._
+import org.apache.calcite.rel.logical.LogicalJoin
+import org.apache.calcite.rel.logical.LogicalUnion
+import org.apache.calcite.rel.core.Join
+import org.apache.calcite.rel.core.Union
+
+/**
+ * This rule is a copy of Calcite's JoinUnionTransposeRule.
+ * Calcite's implementation checks whether one of the operands is a LogicalUnion,
+ * which fails in our case, when it matches with a DataSetUnion.
+ * This rule changes this check to match Union, instead of LogicalUnion only.
+ * The rest of the rule's logic has not been changed.
+ */
+class FlinkJoinUnionTransposeRule(
+ operand: RelOptRuleOperand,
+ description: String) extends RelOptRule(operand, description) {
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+ val join = call.rel(0).asInstanceOf[Join]
+ val (unionRel: Union, otherInput: RelNode, unionOnLeft: Boolean) = {
+ if (call.rel(1).isInstanceOf[Union]) {
+ (call.rel(1).asInstanceOf[Union], call.rel(2).asInstanceOf[RelNode], true)
+ }
+ else {
+ (call.rel(2).asInstanceOf[Union], call.rel(1).asInstanceOf[RelNode], false)
+ }
+ }
+
+ if (!unionRel.all) {
+ return
+ }
+ if (!join.getVariablesStopped.isEmpty) {
+ return
+ }
+ // The UNION ALL cannot be on the null generating side
+ // of an outer join (otherwise we might generate incorrect
+ // rows for the other side for join keys which lack a match
+ // in one or both branches of the union)
+ if (unionOnLeft) {
+ if (join.getJoinType.generatesNullsOnLeft) {
+ return
+ }
+ }
+ else {
+ if (join.getJoinType.generatesNullsOnRight) {
+ return
+ }
+ }
+ val newUnionInputs = new ArrayList[RelNode]
+ for (input <- unionRel.getInputs) {
+ val (joinLeft: RelNode, joinRight: RelNode) = {
+ if (unionOnLeft) {
+ (input, otherInput)
+ }
+ else {
+ (otherInput, input)
+ }
+ }
+
+ newUnionInputs.add(
+ join.copy(
+ join.getTraitSet,
+ join.getCondition,
+ joinLeft,
+ joinRight,
+ join.getJoinType,
+ join.isSemiJoinDone))
+ }
+ val newUnionRel = unionRel.copy(unionRel.getTraitSet, newUnionInputs, true)
+ call.transformTo(newUnionRel)
+ }
+}
+
+object FlinkJoinUnionTransposeRule {
+ val LEFT_UNION = new FlinkJoinUnionTransposeRule(
+ operand(classOf[LogicalJoin], operand(classOf[LogicalUnion], any),
+ operand(classOf[RelNode], any)),
+ "JoinUnionTransposeRule(Union-Other)")
+
+ val RIGHT_UNION = new FlinkJoinUnionTransposeRule(
+ operand(classOf[LogicalJoin], operand(classOf[RelNode], any),
+ operand(classOf[LogicalUnion], any)),
+ "JoinUnionTransposeRule(Other-Union)")
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala
deleted file mode 100644
index 01a5130..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.rules.logical
-
-import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.convert.ConverterRule
-import org.apache.calcite.rel.logical.LogicalAggregate
-import org.apache.flink.api.table.plan.nodes.dataset.{DataSetAggregate, DataSetConvention}
-import scala.collection.JavaConversions._
-
-class FlinkAggregateRule
- extends ConverterRule(
- classOf[LogicalAggregate],
- Convention.NONE,
- DataSetConvention.INSTANCE,
- "FlinkAggregateRule")
- {
-
- def convert(rel: RelNode): RelNode = {
- val agg: LogicalAggregate = rel.asInstanceOf[LogicalAggregate]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
- val convInput: RelNode = RelOptRule.convert(agg.getInput, DataSetConvention.INSTANCE)
-
- new DataSetAggregate(
- rel.getCluster,
- traitSet,
- convInput,
- agg.getNamedAggCalls,
- rel.getRowType,
- agg.getInput.getRowType,
- agg.toString,
- agg.getGroupSet.toArray)
- }
- }
-
-object FlinkAggregateRule {
- val INSTANCE: RelOptRule = new FlinkAggregateRule
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
deleted file mode 100644
index f5e9c68..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.rules.logical
-
-import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.convert.ConverterRule
-import org.apache.calcite.rel.logical.LogicalCalc
-import org.apache.flink.api.table.plan.nodes.dataset.{DataSetCalc, DataSetConvention}
-
-class FlinkCalcRule
- extends ConverterRule(
- classOf[LogicalCalc],
- Convention.NONE,
- DataSetConvention.INSTANCE,
- "FlinkCalcRule")
- {
-
- def convert(rel: RelNode): RelNode = {
- val calc: LogicalCalc = rel.asInstanceOf[LogicalCalc]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
- val convInput: RelNode = RelOptRule.convert(calc.getInput, DataSetConvention.INSTANCE)
-
- new DataSetCalc(
- rel.getCluster,
- traitSet,
- convInput,
- rel.getRowType,
- calc.getProgram,
- calc.toString,
- description)
- }
- }
-
-object FlinkCalcRule {
- val INSTANCE: RelOptRule = new FlinkCalcRule
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
deleted file mode 100644
index c8ce944..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.rules.logical
-
-import org.apache.calcite.plan.{RelOptRuleCall, Convention, RelOptRule, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.convert.ConverterRule
-import org.apache.calcite.rel.logical.LogicalJoin
-import org.apache.calcite.rex.{RexInputRef, RexCall}
-import org.apache.calcite.sql.fun.SqlStdOperatorTable
-import org.apache.flink.api.java.operators.join.JoinType
-import org.apache.flink.api.table.plan.nodes.dataset.{DataSetJoin, DataSetConvention}
-import scala.collection.JavaConverters._
-import scala.collection.JavaConversions._
-
-class FlinkJoinRule
- extends ConverterRule(
- classOf[LogicalJoin],
- Convention.NONE,
- DataSetConvention.INSTANCE,
- "FlinkJoinRule")
- {
-
- override def matches(call: RelOptRuleCall): Boolean = {
-
- val join = call.rel(0).asInstanceOf[LogicalJoin]
- val children = join.getInputs
- val rexBuilder = call.builder().getRexBuilder
-
- val joinInfo = join.analyzeCondition()
- val joinCondition = join.getCondition
- val equiCondition =
- joinInfo.getEquiCondition(children.get(0), children.get(1), rexBuilder)
-
- // joins require at least one equi-condition
- if (equiCondition.isAlwaysTrue) {
- false
- }
- else {
- // check that all equality predicates refer to field refs only (not computed expressions)
- // Note: Calcite treats equality predicates on expressions as non-equi predicates
- joinCondition match {
-
- // conjunction of join predicates
- case c: RexCall if c.getOperator.equals(SqlStdOperatorTable.AND) =>
-
- c.getOperands.asScala
- // look at equality predicates only
- .filter { o =>
- o.isInstanceOf[RexCall] &&
- o.asInstanceOf[RexCall].getOperator.equals(SqlStdOperatorTable.EQUALS)
- }
- // check that both children are field references
- .map { o =>
- o.asInstanceOf[RexCall].getOperands.get(0).isInstanceOf[RexInputRef] &&
- o.asInstanceOf[RexCall].getOperands.get(1).isInstanceOf[RexInputRef]
- }
- // any equality predicate that does not refer to a field reference?
- .reduce( (a, b) => a && b)
-
- // single equi-join predicate
- case c: RexCall if c.getOperator.equals(SqlStdOperatorTable.EQUALS) =>
- c.getOperands.get(0).isInstanceOf[RexInputRef] &&
- c.getOperands.get(1).isInstanceOf[RexInputRef]
- case _ =>
- false
- }
- }
-
- }
-
- def convert(rel: RelNode): RelNode = {
-
- val join: LogicalJoin = rel.asInstanceOf[LogicalJoin]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
- val convLeft: RelNode = RelOptRule.convert(join.getInput(0), DataSetConvention.INSTANCE)
- val convRight: RelNode = RelOptRule.convert(join.getInput(1), DataSetConvention.INSTANCE)
- val joinInfo = join.analyzeCondition
-
- new DataSetJoin(
- rel.getCluster,
- traitSet,
- convLeft,
- convRight,
- rel.getRowType,
- join.toString,
- join.getCondition,
- join.getRowType,
- joinInfo,
- joinInfo.pairs.toList,
- JoinType.INNER,
- null,
- description)
- }
- }
-
-object FlinkJoinRule {
- val INSTANCE: RelOptRule = new FlinkJoinRule
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinUnionTransposeRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinUnionTransposeRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinUnionTransposeRule.scala
deleted file mode 100644
index ff3ee8f..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinUnionTransposeRule.scala
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.rules.logical
-
-import org.apache.calcite.plan.RelOptRule.{any, operand, convert => convertTrait}
-import org.apache.calcite.plan.RelOptRule
-import org.apache.calcite.plan.RelOptRuleOperand
-import org.apache.calcite.plan.RelOptRuleCall
-import org.apache.calcite.rel.RelNode
-import java.util.ArrayList
-import scala.collection.JavaConversions._
-import org.apache.calcite.rel.logical.LogicalJoin
-import org.apache.calcite.rel.logical.LogicalUnion
-import org.apache.calcite.rel.core.Join
-import org.apache.calcite.rel.core.Union
-
-/**
- * This rule is a copy of Calcite's JoinUnionTransposeRule.
- * Calcite's implementation checks whether one of the operands is a LogicalUnion,
- * which fails in our case, when it matches with a DataSetUnion.
- * This rule changes this check to match Union, instead of LogicalUnion only.
- * The rest of the rule's logic has not been changed.
- */
-class FlinkJoinUnionTransposeRule(
- operand: RelOptRuleOperand,
- description: String) extends RelOptRule(operand, description) {
-
- override def onMatch(call: RelOptRuleCall): Unit = {
- val join = call.rel(0).asInstanceOf[Join]
- val (unionRel: Union, otherInput: RelNode, unionOnLeft: Boolean) = {
- if (call.rel(1).isInstanceOf[Union]) {
- (call.rel(1).asInstanceOf[Union], call.rel(2).asInstanceOf[RelNode], true)
- }
- else {
- (call.rel(2).asInstanceOf[Union], call.rel(1).asInstanceOf[RelNode], false)
- }
- }
-
- if (!unionRel.all) {
- return
- }
- if (!join.getVariablesStopped.isEmpty) {
- return
- }
- // The UNION ALL cannot be on the null generating side
- // of an outer join (otherwise we might generate incorrect
- // rows for the other side for join keys which lack a match
- // in one or both branches of the union)
- if (unionOnLeft) {
- if (join.getJoinType.generatesNullsOnLeft) {
- return
- }
- }
- else {
- if (join.getJoinType.generatesNullsOnRight) {
- return
- }
- }
- val newUnionInputs = new ArrayList[RelNode]
- for (input <- unionRel.getInputs) {
- val (joinLeft: RelNode, joinRight: RelNode) = {
- if (unionOnLeft) {
- (input, otherInput)
- }
- else {
- (otherInput, input)
- }
- }
-
- newUnionInputs.add(
- join.copy(
- join.getTraitSet,
- join.getCondition,
- joinLeft,
- joinRight,
- join.getJoinType,
- join.isSemiJoinDone))
- }
- val newUnionRel = unionRel.copy(unionRel.getTraitSet, newUnionInputs, true)
- call.transformTo(newUnionRel)
- }
-}
-
-object FlinkJoinUnionTransposeRule {
- val LEFT_UNION = new FlinkJoinUnionTransposeRule(
- operand(classOf[LogicalJoin], operand(classOf[LogicalUnion], any),
- operand(classOf[RelNode], any)),
- "JoinUnionTransposeRule(Union-Other)")
-
- val RIGHT_UNION = new FlinkJoinUnionTransposeRule(
- operand(classOf[LogicalJoin], operand(classOf[RelNode], any),
- operand(classOf[LogicalUnion], any)),
- "JoinUnionTransposeRule(Other-Union)")
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala
deleted file mode 100644
index 21da504..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.rules.logical
-
-import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.convert.ConverterRule
-import org.apache.calcite.rel.core.TableScan
-import org.apache.calcite.rel.logical.LogicalTableScan
-import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetSource}
-import org.apache.flink.api.table.plan.schema.DataSetTable
-
-class FlinkScanRule
- extends ConverterRule(
- classOf[LogicalTableScan],
- Convention.NONE,
- DataSetConvention.INSTANCE,
- "FlinkScanRule")
- {
- def convert(rel: RelNode): RelNode = {
- val scan: TableScan = rel.asInstanceOf[TableScan]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
-
- new DataSetSource(
- rel.getCluster,
- traitSet,
- scan.getTable,
- rel.getRowType
- )
- }
- }
-
-object FlinkScanRule {
- val INSTANCE: RelOptRule = new FlinkScanRule
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
deleted file mode 100644
index 11600a2..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.rules.logical
-
-import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.convert.ConverterRule
-import org.apache.calcite.rel.logical.LogicalUnion
-import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetUnion}
-
-class FlinkUnionRule
- extends ConverterRule(
- classOf[LogicalUnion],
- Convention.NONE,
- DataSetConvention.INSTANCE,
- "FlinkUnionRule")
- {
-
- def convert(rel: RelNode): RelNode = {
-
- val union: LogicalUnion = rel.asInstanceOf[LogicalUnion]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
- val convLeft: RelNode = RelOptRule.convert(union.getInput(0), DataSetConvention.INSTANCE)
- val convRight: RelNode = RelOptRule.convert(union.getInput(1), DataSetConvention.INSTANCE)
-
- new DataSetUnion(
- rel.getCluster,
- traitSet,
- convLeft,
- convRight,
- rel.getRowType,
- union.toString)
- }
- }
-
-object FlinkUnionRule {
- val INSTANCE: RelOptRule = new FlinkUnionRule
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala
index 11857df..70d0497 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala
@@ -26,7 +26,6 @@ import org.apache.calcite.sql.`type`.SqlTypeName._
import org.apache.calcite.sql.`type`.{SqlTypeFactoryImpl, SqlTypeName}
import org.apache.calcite.sql.fun._
import org.apache.flink.api.common.functions.{GroupReduceFunction, MapFunction}
-import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.table.plan.{TypeConverter, PlanGenException}
import org.apache.flink.api.table.plan.TypeConverter._
import org.apache.flink.api.table.typeinfo.RowTypeInfo
@@ -66,7 +65,7 @@ object AggregateUtil {
def createOperatorFunctionsForAggregates(namedAggregates: Seq[CalcitePair[AggregateCall, String]],
inputType: RelDataType, outputType: RelDataType,
groupings: Array[Int],
- config: TableConfig): AggregateResult = {
+ config: TableConfig): (MapFunction[Any, Row], GroupReduceFunction[Row, Row] ) = {
val aggregateFunctionsAndFieldIndexes =
transformToAggregateFunctions(namedAggregates.map(_.getKey), inputType, groupings.length)
@@ -85,7 +84,7 @@ object AggregateUtil {
val mapFunction = new AggregateMapFunction[Row, Row](
aggregates, aggFieldIndexes, groupings,
- mapReturnType.asInstanceOf[RowTypeInfo]).asInstanceOf[MapFunction[Any, Any]]
+ mapReturnType.asInstanceOf[RowTypeInfo]).asInstanceOf[MapFunction[Any, Row]]
// the mapping relation between field index of intermediate aggregate Row and output Row.
val groupingOffsetMapping = getGroupKeysMapping(inputType, outputType, groupings)
@@ -114,7 +113,7 @@ object AggregateUtil {
aggOffsetMapping, intermediateRowArity)
}
- new AggregateResult(mapFunction, reduceGroupFunction)
+ (mapFunction, reduceGroupFunction)
}
private def transformToAggregateFunctions(
@@ -317,7 +316,3 @@ object AggregateUtil {
}
}
-case class AggregateResult(
- val mapFunc: MapFunction[Any, Any],
- val reduceGroupFunc: GroupReduceFunction[Row, Row]) {
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d720b002/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala.orig
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala.orig b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala.orig
deleted file mode 100644
index ad6a641..0000000
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala.orig
+++ /dev/null
@@ -1,164 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.scala.table.test
-
-import org.apache.flink.api.table.Row
-import org.apache.flink.api.scala._
-import org.apache.flink.api.scala.table._
-import org.apache.flink.api.scala.util.CollectionDataSets
-import org.apache.flink.api.table.expressions.Literal
-import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
-import org.junit._
-import org.junit.runner.RunWith
-import org.junit.runners.Parameterized
-
-import scala.collection.JavaConverters._
-
-@RunWith(classOf[Parameterized])
-class GroupedAggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
-
- @Test(expected = classOf[IllegalArgumentException])
- def testGroupingOnNonExistentField(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
- // must fail. '_foo not a valid field
- .groupBy('_foo)
- .select('a.avg)
- }
-
- @Test(expected = classOf[IllegalArgumentException])
- def testGroupingInvalidSelection(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
- .groupBy('a, 'b)
- // must fail. 'c is not a grouping key or aggregation
- .select('c)
- }
-
- @Test
- def testGroupedAggregate(): Unit = {
-
- // the grouping key needs to be forwarded to the intermediate DataSet, even
- // if we don't want the key in the output
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
- .groupBy('b)
- .select('b, 'a.sum)
-
- val expected = "1,1\n" + "2,5\n" + "3,15\n" + "4,34\n" + "5,65\n" + "6,111\n"
- val results = t.toDataSet[Row].collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testGroupingKeyForwardIfNotUsed(): Unit = {
-
- // the grouping key needs to be forwarded to the intermediate DataSet, even
- // if we don't want the key in the output
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
- .groupBy('b)
- .select('a.sum)
-
- val expected = "1\n" + "5\n" + "15\n" + "34\n" + "65\n" + "111\n"
- val results = t.toDataSet[Row].collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testGroupNoAggregation(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val t = CollectionDataSets.get3TupleDataSet(env)
- .as('a, 'b, 'c)
- .groupBy('b)
- .select('a.sum as 'd, 'b)
- .groupBy('b, 'd)
- .select('b)
-
- val expected = "1\n" + "2\n" + "3\n" + "4\n" + "5\n" + "6\n"
- val results = t.toDataSet[Row].collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testGroupedAggregateWithLongKeys(): Unit = {
- // This uses very long keys to force serialized comparison.
- // With short keys, the normalized key is sufficient.
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements(
- ("hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhaa", 1, 2),
- ("hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhaa", 1, 2),
- ("hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhaa", 1, 2),
- ("hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhaa", 1, 2),
- ("hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhaa", 1, 2),
- ("hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhab", 1, 2),
- ("hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhab", 1, 2),
- ("hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhab", 1, 2),
- ("hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhab", 1, 2))
- .rebalance().setParallelism(2).as('a, 'b, 'c)
- .groupBy('a, 'b)
- .select('c.sum)
-
- val expected = "10\n" + "8\n"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testGroupedByExpression(): Unit = {
-
- // verify AggregateProjectPullUpConstantsRule
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
- .select('a, 4 as 'four, 'b)
- .groupBy('four, 'a)
- .select('four, 'b.sum)
-
- val expected = "4,2\n" + "4,3\n" + "4,5\n" + "4,5\n" + "4,5\n" + "4,6\n" +
- "4,6\n" + "4,6\n" + "4,3\n" + "4,4\n" + "4,6\n" + "4,1\n" + "4,4\n" +
- "4,4\n" + "4,5\n" + "4,6\n" + "4,2\n" + "4,3\n" + "4,4\n" + "4,5\n" + "4,6\n"
- val results = t.toDataSet[Row].collect()
-
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testGroupedByExpression2(): Unit = {
-
- // verify AggregateProjectPullUpConstantsRule
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
- .select('b, 4 as 'four, 'a)
- .groupBy('b, 'four)
- .select('four, 'a.sum)
-
- val expected = "4,1\n" + "4,5\n" + "4,15\n" + "4,34\n" + "4,65\n" + "4,111\n"
- val results = t.toDataSet[Row].collect()
-
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-}
[2/3] flink git commit: [FLINK-3596] DataSet RelNode refactoring
Posted by fh...@apache.org.
[FLINK-3596] DataSet RelNode refactoring
- remove the intermediate flink relnode layer and the dataset rules
- move code generation from rules to DataSet nodes
- remove unused DataSete nodes
- move code generation from join rule to DataSetJoin node
- merge DataSetMap and DataSetReduce into DataSetAggregate
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/22621e02
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/22621e02
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/22621e02
Branch: refs/heads/tableOnCalcite
Commit: 22621e0289ee6e2783b3220685d9aaba84476d77
Parents: e34e439
Author: vasia <va...@apache.org>
Authored: Tue Mar 8 17:29:32 2016 +0100
Committer: Fabian Hueske <fh...@apache.org>
Committed: Thu Mar 10 23:30:26 2016 +0100
----------------------------------------------------------------------
.../api/java/table/JavaBatchTranslator.scala | 28 +---
.../plan/nodes/dataset/DataSetAggregate.scala | 114 ++++++++++++++
.../table/plan/nodes/dataset/DataSetCalc.scala | 152 +++++++++++++++++++
.../plan/nodes/dataset/DataSetExchange.scala | 66 --------
.../plan/nodes/dataset/DataSetFlatMap.scala | 74 ---------
.../plan/nodes/dataset/DataSetGroupReduce.scala | 105 -------------
.../table/plan/nodes/dataset/DataSetJoin.scala | 83 ++++++++--
.../table/plan/nodes/dataset/DataSetMap.scala | 90 -----------
.../plan/nodes/dataset/DataSetReduce.scala | 66 --------
.../table/plan/nodes/dataset/DataSetSort.scala | 65 --------
.../plan/nodes/logical/FlinkAggregate.scala | 60 --------
.../table/plan/nodes/logical/FlinkCalc.scala | 37 -----
.../plan/nodes/logical/FlinkConvention.scala | 42 -----
.../table/plan/nodes/logical/FlinkJoin.scala | 46 ------
.../api/table/plan/nodes/logical/FlinkRel.scala | 25 ---
.../table/plan/nodes/logical/FlinkScan.scala | 31 ----
.../table/plan/nodes/logical/FlinkUnion.scala | 38 -----
.../api/table/plan/rules/FlinkRuleSets.scala | 11 --
.../rules/dataset/DataSetAggregateRule.scala | 71 ---------
.../plan/rules/dataset/DataSetCalcRule.scala | 137 -----------------
.../plan/rules/dataset/DataSetJoinRule.scala | 135 ----------------
.../plan/rules/dataset/DataSetScanRule.scala | 51 -------
.../plan/rules/dataset/DataSetUnionRule.scala | 53 -------
.../plan/rules/logical/FlinkAggregateRule.scala | 22 +--
.../plan/rules/logical/FlinkCalcRule.scala | 15 +-
.../plan/rules/logical/FlinkJoinRule.scala | 37 +++--
.../logical/FlinkJoinUnionTransposeRule.scala | 2 +-
.../plan/rules/logical/FlinkScanRule.scala | 11 +-
.../plan/rules/logical/FlinkUnionRule.scala | 22 +--
.../table/runtime/aggregate/AggregateUtil.scala | 49 +++---
30 files changed, 421 insertions(+), 1317 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
index f238df3..14ee78e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
@@ -76,8 +76,10 @@ class JavaBatchTranslator(config: TableConfig) extends PlanTranslator {
// optimize the logical Flink plan
val optProgram = Programs.ofRules(FlinkRuleSets.DATASET_OPT_RULES)
val flinkOutputProps = RelTraitSet.createEmpty()
+ .plus(DataSetConvention.INSTANCE)
+ .plus(RelCollations.of()).simplify()
- val optPlan = try {
+ val dataSetPlan = try {
optProgram.run(planner, decorPlan, flinkOutputProps)
}
catch {
@@ -89,30 +91,8 @@ class JavaBatchTranslator(config: TableConfig) extends PlanTranslator {
}
println("---------------")
- println("Optimized Plan:")
- println("---------------")
- println(RelOptUtil.toString(optPlan))
-
- // optimize the logical Flink plan
- val dataSetProgram = Programs.ofRules(FlinkRuleSets.DATASET_TRANS_RULES)
- val dataSetOutputProps = RelTraitSet.createEmpty()
- .plus(DataSetConvention.INSTANCE)
- .plus(RelCollations.of()).simplify()
-
- val dataSetPlan = try {
- dataSetProgram.run(planner, optPlan, dataSetOutputProps)
- }
- catch {
- case e: CannotPlanException =>
- throw new PlanGenException(
- s"Cannot generate a valid execution plan for the given query: \n\n" +
- s"${RelOptUtil.toString(lPlan)}\n" +
- "Please consider filing a bug report.", e)
- }
-
- println("-------------")
println("DataSet Plan:")
- println("-------------")
+ println("---------------")
println(RelOptUtil.toString(dataSetPlan))
dataSetPlan match {
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
new file mode 100644
index 0000000..3856c5f
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.AggregateCall
+import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.plan.TypeConverter._
+import org.apache.flink.api.table.plan.{PlanGenException, TypeConverter}
+import org.apache.flink.api.table.runtime.aggregate.AggregateUtil
+import org.apache.flink.api.table.runtime.aggregate.AggregateUtil.CalcitePair
+import org.apache.flink.api.table.typeinfo.RowTypeInfo
+import org.apache.flink.api.table.{Row, TableConfig}
+
+import scala.collection.JavaConverters._
+
+/**
+ * Flink RelNode which matches along with a LogicalAggregate.
+ */
+class DataSetAggregate(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ namedAggregates: Seq[CalcitePair[AggregateCall, String]],
+ rowType: RelDataType,
+ inputType: RelDataType,
+ opName: String,
+ grouping: Array[Int])
+ extends SingleRel(cluster, traitSet, input)
+ with DataSetRel {
+
+ override def deriveRowType() = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new DataSetAggregate(
+ cluster,
+ traitSet,
+ inputs.get(0),
+ namedAggregates,
+ rowType,
+ inputType,
+ opName,
+ grouping)
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw).item("name", opName)
+ }
+
+ override def translateToPlan(
+ config: TableConfig,
+ expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
+
+ expectedType match {
+ case Some(typeInfo) if typeInfo.getTypeClass != classOf[Row] =>
+ throw new PlanGenException("Aggregate operations currently only support returning Rows.")
+ case _ => // ok
+ }
+
+ val groupingKeys = (0 until grouping.length).toArray
+ // add grouping fields, position keys in the input, and input type
+ val aggregateResult = AggregateUtil.createOperatorFunctionsForAggregates(namedAggregates,
+ inputType, rowType, grouping, config)
+
+ val inputDS = input.asInstanceOf[DataSetRel].translateToPlan(
+ config,
+ // tell the input operator that this operator currently only supports Rows as input
+ Some(TypeConverter.DEFAULT_ROW_TYPE))
+
+ // get the output types
+ val fieldTypes: Array[TypeInformation[_]] = rowType.getFieldList.asScala
+ .map(f => f.getType.getSqlTypeName)
+ .map(n => TypeConverter.sqlTypeToTypeInfo(n))
+ .toArray
+
+ val rowTypeInfo = new RowTypeInfo(fieldTypes)
+ val mappedInput = inputDS.map(aggregateResult.mapFunc)
+ val groupReduceFunction = aggregateResult.reduceGroupFunc
+
+ if (groupingKeys.length > 0) {
+ mappedInput.asInstanceOf[DataSet[Row]]
+ .groupBy(groupingKeys: _*)
+ .reduceGroup(groupReduceFunction)
+ .returns(rowTypeInfo)
+ .asInstanceOf[DataSet[Any]]
+ }
+ else {
+ // global aggregation
+ mappedInput.asInstanceOf[DataSet[Row]]
+ .reduceGroup(groupReduceFunction)
+ .returns(rowTypeInfo)
+ .asInstanceOf[DataSet[Any]]
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetCalc.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetCalc.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetCalc.scala
new file mode 100644
index 0000000..d7c71cc
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetCalc.scala
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
+import org.apache.flink.api.common.functions.FlatMapFunction
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.codegen.CodeGenerator
+import org.apache.flink.api.table.plan.TypeConverter._
+import org.apache.flink.api.table.runtime.FlatMapRunner
+import org.apache.flink.api.table.TableConfig
+import org.apache.calcite.rex.RexProgram
+import scala.collection.JavaConversions._
+
+/**
+ * Flink RelNode which matches along with LogicalCalc.
+ *
+ */
+class DataSetCalc(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ rowType: RelDataType,
+ calcProgram: RexProgram,
+ opName: String,
+ ruleDescription: String)
+ extends SingleRel(cluster, traitSet, input)
+ with DataSetRel {
+
+ override def deriveRowType() = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new DataSetCalc(
+ cluster,
+ traitSet,
+ inputs.get(0),
+ rowType,
+ calcProgram,
+ opName,
+ ruleDescription)
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw).item("name", opName)
+ }
+
+ override def toString = opName
+
+ override def translateToPlan(config: TableConfig,
+ expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
+
+ val inputDS = input.asInstanceOf[DataSetRel].translateToPlan(config)
+
+ val returnType = determineReturnType(
+ getRowType,
+ expectedType,
+ config.getNullCheck,
+ config.getEfficientTypeUsage)
+
+ val generator = new CodeGenerator(config, inputDS.getType)
+
+ val condition = calcProgram.getCondition
+ val expandedExpressions = calcProgram.getProjectList.map(
+ expr => calcProgram.expandLocalRef(expr))
+ val projection = generator.generateResultExpression(
+ returnType,
+ rowType.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 (inputDS.getType != returnType) {
+ val conversion = generator.generateConverterResultExpression(
+ returnType,
+ rowType.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(
+ ruleDescription,
+ classOf[FlatMapFunction[Any, Any]],
+ body,
+ returnType)
+
+ val mapFunc = new FlatMapRunner[Any, Any](
+ genFunction.name,
+ genFunction.code,
+ genFunction.returnType)
+
+ inputDS.flatMap(mapFunc)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
deleted file mode 100644
index 00cf899..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.dataset
-
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
-import org.apache.flink.api.common.operators.base.PartitionOperatorBase.PartitionMethod
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.TableConfig
-
-/**
- * Flink RelNode which matches along with PartitionOperator.
- */
-class DataSetExchange(
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- input: RelNode,
- rowType: RelDataType,
- opName: String,
- partitionKey: Array[Int],
- partitionMethod: PartitionMethod)
- extends SingleRel(cluster, traitSet, input)
- with DataSetRel {
-
- override def deriveRowType() = rowType
-
- override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
- new DataSetExchange(
- cluster,
- traitSet,
- inputs.get(0),
- rowType,
- opName,
- partitionKey,
- partitionMethod
- )
- }
-
- override def explainTerms(pw: RelWriter): RelWriter = {
- super.explainTerms(pw).item("name", opName)
- }
-
- override def translateToPlan(
- config: TableConfig,
- expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
- ???
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
deleted file mode 100644
index 9744792..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.dataset
-
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
-import org.apache.flink.api.common.functions.FlatMapFunction
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.TableConfig
-import org.apache.flink.api.table.plan.TypeConverter._
-
-/**
- * Flink RelNode which matches along with FlatMapOperator.
- *
- */
-class DataSetFlatMap(
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- input: RelNode,
- rowType: RelDataType,
- opName: String,
- func: (TableConfig, TypeInformation[Any], TypeInformation[Any]) => FlatMapFunction[Any, Any])
- extends SingleRel(cluster, traitSet, input)
- with DataSetRel {
-
- override def deriveRowType() = rowType
-
- override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
- new DataSetFlatMap(
- cluster,
- traitSet,
- inputs.get(0),
- rowType,
- opName,
- func
- )
- }
-
- override def explainTerms(pw: RelWriter): RelWriter = {
- super.explainTerms(pw).item("name", opName)
- }
-
- override def toString = opName
-
- override def translateToPlan(config: TableConfig,
- expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
- val inputDataSet = input.asInstanceOf[DataSetRel].translateToPlan(config)
- val returnType = determineReturnType(
- getRowType,
- expectedType,
- config.getNullCheck,
- config.getEfficientTypeUsage)
- val flatMapFunc = func.apply(config, inputDataSet.getType, returnType)
- inputDataSet.flatMap(flatMapFunc)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
deleted file mode 100644
index b87a092..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.dataset
-
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
-import org.apache.flink.api.common.functions.GroupReduceFunction
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.plan.{PlanGenException, TypeConverter}
-import org.apache.flink.api.table.typeinfo.RowTypeInfo
-import org.apache.flink.api.table.{Row, TableConfig}
-
-import scala.collection.JavaConverters._
-
-/**
- * Flink RelNode which matches along with ReduceGroupOperator.
- */
-class DataSetGroupReduce(
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- input: RelNode,
- rowType: RelDataType,
- opName: String,
- groupingKeys: Array[Int],
- func: (TableConfig, TypeInformation[Row], TypeInformation[Row]) =>
- GroupReduceFunction[Row, Row])
- extends SingleRel(cluster, traitSet, input)
- with DataSetRel {
-
- override def deriveRowType() = rowType
-
- override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
- new DataSetGroupReduce(
- cluster,
- traitSet,
- inputs.get(0),
- rowType,
- opName,
- groupingKeys,
- func
- )
- }
-
- override def explainTerms(pw: RelWriter): RelWriter = {
- super.explainTerms(pw).item("name", opName)
- }
-
- override def translateToPlan(
- config: TableConfig,
- expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
-
- expectedType match {
- case Some(typeInfo) if typeInfo.getTypeClass != classOf[Row] =>
- throw new PlanGenException("GroupReduce operations currently only support returning Rows.")
- case _ => // ok
- }
-
- val inputDS = input.asInstanceOf[DataSetRel].translateToPlan(
- config,
- // tell the input operator that this operator currently only supports Rows as input
- Some(TypeConverter.DEFAULT_ROW_TYPE))
-
- // get the output types
- val fieldsNames = rowType.getFieldNames
- val fieldTypes: Array[TypeInformation[_]] = rowType.getFieldList.asScala
- .map(f => f.getType.getSqlTypeName)
- .map(n => TypeConverter.sqlTypeToTypeInfo(n))
- .toArray
-
- val rowTypeInfo = new RowTypeInfo(fieldTypes)
- val groupReduceFunction =
- func.apply(config, inputDS.getType.asInstanceOf[RowTypeInfo], rowTypeInfo)
-
- if (groupingKeys.length > 0) {
- inputDS.asInstanceOf[DataSet[Row]]
- .groupBy(groupingKeys: _*)
- .reduceGroup(groupReduceFunction)
- .returns(rowTypeInfo)
- .asInstanceOf[DataSet[Any]]
- }
- else {
- // global aggregation
- inputDS.asInstanceOf[DataSet[Row]].reduceGroup(groupReduceFunction)
- .returns(rowTypeInfo).asInstanceOf[DataSet[Any]]
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
index c32853d..1d293d2 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
@@ -20,21 +20,21 @@ package org.apache.flink.api.table.plan.nodes.dataset
import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.JoinInfo
import org.apache.calcite.rel.{RelWriter, BiRel, RelNode}
-import org.apache.flink.api.common.functions.JoinFunction
+import org.apache.calcite.util.mapping.IntPair
import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.DataSet
import org.apache.flink.api.java.operators.join.JoinType
-import org.apache.flink.api.table.{TableConfig, Row}
+import org.apache.flink.api.table.codegen.CodeGenerator
+import org.apache.flink.api.table.runtime.FlatJoinRunner
+import org.apache.flink.api.table.{TableException, TableConfig}
import org.apache.flink.api.common.functions.FlatJoinFunction
import org.apache.flink.api.table.plan.TypeConverter._
-import org.apache.flink.api.common.functions.MapFunction
-import org.apache.flink.api.java.tuple.Tuple2
-import org.apache.flink.api.table.typeinfo.RowTypeInfo
import scala.collection.mutable.ArrayBuffer
import scala.collection.JavaConversions._
-import org.apache.flink.api.table.plan.TypeConverter
+import org.apache.calcite.rex.RexNode
/**
* Flink RelNode which matches along with JoinOperator and its related operations.
@@ -46,12 +46,13 @@ class DataSetJoin(
right: RelNode,
rowType: RelDataType,
opName: String,
- joinKeysLeft: Array[Int],
- joinKeysRight: Array[Int],
+ joinCondition: RexNode,
+ joinRowType: RelDataType,
+ joinInfo: JoinInfo,
+ keyPairs: List[IntPair],
joinType: JoinType,
joinHint: JoinHint,
- func: (TableConfig, TypeInformation[Any], TypeInformation[Any], TypeInformation[Any]) =>
- FlatJoinFunction[Any, Any, Any])
+ ruleDescription: String)
extends BiRel(cluster, traitSet, left, right)
with DataSetRel {
@@ -65,12 +66,13 @@ class DataSetJoin(
inputs.get(1),
rowType,
opName,
- joinKeysLeft,
- joinKeysRight,
+ joinCondition,
+ joinRowType,
+ joinInfo,
+ keyPairs,
joinType,
joinHint,
- func
- )
+ ruleDescription)
}
override def explainTerms(pw: RelWriter): RelWriter = {
@@ -90,8 +92,57 @@ class DataSetJoin(
config.getNullCheck,
config.getEfficientTypeUsage)
- val joinFun = func.apply(config, leftDataSet.getType, rightDataSet.getType, returnType)
- leftDataSet.join(rightDataSet).where(joinKeysLeft: _*).equalTo(joinKeysRight: _*)
+ // get the equality keys
+ val leftKeys = ArrayBuffer.empty[Int]
+ val rightKeys = ArrayBuffer.empty[Int]
+ if (keyPairs.isEmpty) {
+ // if no equality keys => not supported
+ throw new TableException("Joins should have at least one equality condition")
+ }
+ else {
+ // at least one equality expression => generate a join function
+ keyPairs.foreach(pair => {
+ leftKeys.add(pair.source)
+ rightKeys.add(pair.target)
+ })
+ }
+
+ val generator = new CodeGenerator(config, leftDataSet.getType, Some(rightDataSet.getType))
+ val conversion = generator.generateConverterResultExpression(
+ returnType,
+ joinRowType.getFieldNames)
+
+ var body = ""
+
+ if (joinInfo.isEqui) {
+ // only equality condition
+ body = s"""
+ |${conversion.code}
+ |${generator.collectorTerm}.collect(${conversion.resultTerm});
+ |""".stripMargin
+ }
+ else {
+ val condition = generator.generateExpression(joinCondition)
+ body = s"""
+ |${condition.code}
+ |if (${condition.resultTerm}) {
+ | ${conversion.code}
+ | ${generator.collectorTerm}.collect(${conversion.resultTerm});
+ |}
+ |""".stripMargin
+ }
+ val genFunction = generator.generateFunction(
+ ruleDescription,
+ classOf[FlatJoinFunction[Any, Any, Any]],
+ body,
+ returnType)
+
+ val joinFun = new FlatJoinRunner[Any, Any, Any](
+ genFunction.name,
+ genFunction.code,
+ genFunction.returnType)
+
+ leftDataSet.join(rightDataSet).where(leftKeys.toArray: _*).equalTo(rightKeys.toArray: _*)
.`with`(joinFun).asInstanceOf[DataSet[Any]]
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
deleted file mode 100644
index d87e047..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.dataset
-
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
-import org.apache.flink.api.common.functions.{MapFunction, MapPartitionFunction}
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.plan.{PlanGenException, TypeConverter}
-import org.apache.flink.api.table.plan.TypeConverter._
-import org.apache.flink.api.table.typeinfo.RowTypeInfo
-import org.apache.flink.api.table.{Row, TableConfig}
-
-import scala.collection.JavaConverters._
-
-/**
- * Flink RelNode which matches along with MapOperator.
- *
- */
-class DataSetMap(
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- input: RelNode,
- rowType: RelDataType,
- opName: String,
- func: (TableConfig, TypeInformation[Any], TypeInformation[Any]) => MapFunction[Any, Any])
- extends SingleRel(cluster, traitSet, input)
- with DataSetRel {
-
- override def deriveRowType() = rowType
-
- override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
- new DataSetMap(
- cluster,
- traitSet,
- inputs.get(0),
- rowType,
- opName,
- func
- )
- }
-
- override def explainTerms(pw: RelWriter): RelWriter = {
- super.explainTerms(pw).item("name", opName)
- }
-
- override def toString = opName
-
- override def translateToPlan(config: TableConfig,
- expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
-
- expectedType match {
- case Some(typeInfo) if typeInfo.getTypeClass != classOf[Row] =>
- throw new PlanGenException("GroupReduce operations " +
- "currently only support returning Rows.")
- case _ => // ok
- }
-
- val inputDS = input.asInstanceOf[DataSetRel].translateToPlan(
- config,
- // tell the input operator that this operator currently only supports Rows as input
- Some(TypeConverter.DEFAULT_ROW_TYPE))
-
- val returnType = determineReturnType(
- getRowType,
- expectedType,
- config.getNullCheck,
- config.getEfficientTypeUsage)
- val mapFunc = func.apply(config, inputDS.getType, returnType)
- inputDS.map(mapFunc)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
deleted file mode 100644
index 361f869..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.dataset
-
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
-import org.apache.flink.api.common.functions.ReduceFunction
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.{TableConfig, Row}
-
-/**
- * Flink RelNode which matches along with ReduceOperator.
- */
-class DataSetReduce(
- cluster: RelOptCluster,
- traits: RelTraitSet,
- input: RelNode,
- rowType: RelDataType,
- opName: String,
- groupingKeys: Array[Int],
- func: ReduceFunction[Any])
- extends SingleRel(cluster, traits, input)
- with DataSetRel {
-
- override def deriveRowType() = rowType
-
- override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
- new DataSetReduce(
- cluster,
- traitSet,
- inputs.get(0),
- rowType,
- opName,
- groupingKeys,
- func
- )
- }
-
- override def explainTerms(pw: RelWriter): RelWriter = {
- super.explainTerms(pw).item("name", opName)
- }
-
- override def translateToPlan(
- config: TableConfig,
- expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
- ???
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
deleted file mode 100644
index 033711b..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.dataset
-
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.TableConfig
-
-/**
- * Flink RelNode which matches along with SortPartitionOperator.
- */
-class DataSetSort(
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- input: RelNode,
- rowType: RelDataType,
- opName: String,
- sortKey: Array[Int],
- sortOrder: Array[Boolean])
- extends SingleRel(cluster, traitSet, input)
- with DataSetRel {
-
- override def deriveRowType() = rowType
-
- override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
- new DataSetSort(
- cluster,
- traitSet,
- inputs.get(0),
- rowType,
- opName,
- sortKey,
- sortOrder
- )
- }
-
- override def explainTerms(pw: RelWriter): RelWriter = {
- super.explainTerms(pw).item("name", opName)
- }
-
- override def translateToPlan(
- config: TableConfig,
- expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
- ???
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkAggregate.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkAggregate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkAggregate.scala
deleted file mode 100644
index 1fca03a..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkAggregate.scala
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.logical
-
-import java.util
-
-import org.apache.calcite.plan.{RelOptCost, RelOptPlanner, RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.core.{AggregateCall, Aggregate}
-import org.apache.calcite.sql.fun.SqlAvgAggFunction
-import org.apache.calcite.util.ImmutableBitSet
-
-import scala.collection.JavaConversions._
-
-class FlinkAggregate(
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- input: RelNode,
- indicator: Boolean,
- groupSet: ImmutableBitSet,
- groupSets: java.util.List[ImmutableBitSet],
- aggCalls: java.util.List[AggregateCall])
- extends Aggregate(cluster, traitSet, input, indicator, groupSet, groupSets, aggCalls)
- with FlinkRel {
-
- override def copy(
- traitSet: RelTraitSet,
- input: RelNode,
- indicator: Boolean,
- groupSet: ImmutableBitSet,
- groupSets: util.List[ImmutableBitSet],
- aggCalls: util.List[AggregateCall]): Aggregate = {
-
- new FlinkAggregate(
- cluster,
- traitSet,
- input,
- indicator,
- groupSet,
- groupSets,
- aggCalls
- )
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkCalc.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkCalc.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkCalc.scala
deleted file mode 100644
index bcfe8d7..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkCalc.scala
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.logical
-
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.core.Calc
-import org.apache.calcite.rex.RexProgram
-
-class FlinkCalc(
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- input: RelNode,
- program: RexProgram)
- extends Calc(cluster, traitSet, input, program)
- with FlinkRel {
-
- override def copy(traitSet: RelTraitSet, child: RelNode, program: RexProgram): Calc = {
- new FlinkCalc(cluster, traitSet, child, program)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkConvention.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkConvention.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkConvention.scala
deleted file mode 100644
index 80137f2..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkConvention.scala
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.logical
-
-import org.apache.calcite.plan._
-
-class FlinkConvention extends Convention {
-
- override def toString: String = getName
-
- def getInterface: Class[_] = classOf[FlinkRel]
-
- def getName: String = "FLINK"
-
- def getTraitDef: RelTraitDef[_ <: RelTrait] = ConventionTraitDef.INSTANCE
-
- def satisfies(`trait`: RelTrait): Boolean = this eq `trait`
-
- def register(planner: RelOptPlanner): Unit = { }
-
-}
-
-object FlinkConvention {
-
- val INSTANCE = new FlinkConvention
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkJoin.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkJoin.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkJoin.scala
deleted file mode 100644
index 8b04b50..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkJoin.scala
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.logical
-
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.core.{JoinRelType, Join}
-import org.apache.calcite.rex.RexNode
-
-class FlinkJoin(
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- left: RelNode,
- right: RelNode,
- condition: RexNode,
- joinType: JoinRelType,
- variablesStopped: java.util.Set[String])
- extends Join(cluster, traitSet, left, right, condition, joinType, variablesStopped)
- with FlinkRel {
-
- override def copy(
- traitSet: RelTraitSet,
- condition: RexNode,
- left: RelNode,
- right: RelNode,
- joinType: JoinRelType,
- semiJoinDone: Boolean): Join = {
- new FlinkJoin(cluster, traitSet, left, right, condition, joinType, getVariablesStopped)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkRel.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkRel.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkRel.scala
deleted file mode 100644
index 9ebd7e4..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkRel.scala
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.logical
-
-import org.apache.calcite.rel.RelNode
-
-trait FlinkRel extends RelNode {
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkScan.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkScan.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkScan.scala
deleted file mode 100644
index 6d53a75..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkScan.scala
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.logical
-
-import org.apache.calcite.plan.{RelOptTable, RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.core.TableScan
-
-class FlinkScan(
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- table: RelOptTable)
- extends TableScan(cluster, traitSet, table)
- with FlinkRel {
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkUnion.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkUnion.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkUnion.scala
deleted file mode 100644
index fd791d3..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkUnion.scala
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.nodes.logical
-
-import java.util
-
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.core.{SetOp, Union}
-
-class FlinkUnion(
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- inputs: java.util.List[RelNode],
- all: Boolean)
- extends Union(cluster, traitSet, inputs, all)
- with FlinkRel {
-
- override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode], all: Boolean): SetOp = {
- new FlinkUnion(cluster, traitSet, inputs, all)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
index b5c3800..bd128b2 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
@@ -21,7 +21,6 @@ package org.apache.flink.api.table.plan.rules
import org.apache.calcite.rel.rules._
import org.apache.calcite.tools.{RuleSets, RuleSet}
import org.apache.flink.api.table.plan.rules.logical._
-import org.apache.flink.api.table.plan.rules.dataset._
object FlinkRuleSets {
@@ -102,14 +101,4 @@ object FlinkRuleSets {
FlinkUnionRule.INSTANCE
)
- val DATASET_TRANS_RULES: RuleSet = RuleSets.ofList(
-
- // translate to DataSet nodes
- DataSetAggregateRule.INSTANCE,
- DataSetCalcRule.INSTANCE,
- DataSetJoinRule.INSTANCE,
- DataSetScanRule.INSTANCE,
- DataSetUnionRule.INSTANCE
- )
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetAggregateRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetAggregateRule.scala
deleted file mode 100644
index ba77fea..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetAggregateRule.scala
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.rules.dataset
-
-import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.convert.ConverterRule
-import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetGroupReduce, DataSetMap}
-import org.apache.flink.api.table.plan.nodes.logical.{FlinkAggregate, FlinkConvention}
-import org.apache.flink.api.table.runtime.aggregate.AggregateUtil
-
-import scala.collection.JavaConversions._
-
-class DataSetAggregateRule
- extends ConverterRule(
- classOf[FlinkAggregate],
- FlinkConvention.INSTANCE,
- DataSetConvention.INSTANCE,
- "DataSetAggregateRule")
-{
-
- def convert(rel: RelNode): RelNode = {
- val agg: FlinkAggregate = rel.asInstanceOf[FlinkAggregate]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
- val convInput: RelNode = RelOptRule.convert(agg.getInput, DataSetConvention.INSTANCE)
-
- val grouping = agg.getGroupSet.toArray
-
- val inputType = agg.getInput.getRowType()
-
- // add grouping fields, position keys in the input, and input type
- val aggregateResult = AggregateUtil.createOperatorFunctionsForAggregates(agg.getNamedAggCalls,
- inputType, rel.getRowType, grouping)
-
- val mapNode = new DataSetMap(rel.getCluster,
- traitSet,
- convInput,
- aggregateResult.intermediateDataType,
- agg.toString,
- aggregateResult.mapFunc)
-
- new DataSetGroupReduce(
- rel.getCluster,
- traitSet,
- mapNode,
- rel.getRowType,
- agg.toString,
- (0 until grouping.length).toArray,
- aggregateResult.reduceGroupFunc)
- }
-}
-
-object DataSetAggregateRule {
- val INSTANCE: RelOptRule = new DataSetAggregateRule
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala
deleted file mode 100644
index 256a085..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala
+++ /dev/null
@@ -1,137 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.rules.dataset
-
-import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
-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(
- classOf[FlinkCalc],
- FlinkConvention.INSTANCE,
- DataSetConvention.INSTANCE,
- "DataSetCalcRule")
-{
-
- def convert(rel: RelNode): RelNode = {
- val calc: FlinkCalc = rel.asInstanceOf[FlinkCalc]
- 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,
- calcFunc)
- }
-}
-
-object DataSetCalcRule {
- val INSTANCE: RelOptRule = new DataSetCalcRule
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetJoinRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetJoinRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetJoinRule.scala
deleted file mode 100644
index c045471..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetJoinRule.scala
+++ /dev/null
@@ -1,135 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.rules.dataset
-
-import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.convert.ConverterRule
-import org.apache.flink.api.java.operators.join.JoinType
-import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetJoin}
-import org.apache.flink.api.table.plan.nodes.logical.{FlinkJoin, FlinkConvention}
-import scala.collection.JavaConversions._
-import scala.collection.mutable.ArrayBuffer
-import org.apache.flink.api.table.plan.TypeConverter._
-import org.apache.flink.api.table.runtime.FlatJoinRunner
-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.FlatJoinFunction
-import org.apache.calcite.rel.core.JoinInfo
-import org.apache.flink.api.table.TableException
-
-class DataSetJoinRule
- extends ConverterRule(
- classOf[FlinkJoin],
- FlinkConvention.INSTANCE,
- DataSetConvention.INSTANCE,
- "DataSetJoinRule")
-{
-
- def convert(rel: RelNode): RelNode = {
- val join: FlinkJoin = rel.asInstanceOf[FlinkJoin]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
- val convLeft: RelNode = RelOptRule.convert(join.getInput(0), DataSetConvention.INSTANCE)
- val convRight: RelNode = RelOptRule.convert(join.getInput(1), DataSetConvention.INSTANCE)
-
- // get the equality keys
- val joinInfo = join.analyzeCondition
- val keyPairs = joinInfo.pairs
-
- if (keyPairs.isEmpty) { // if no equality keys => not supported
- throw new TableException("Joins should have at least one equality condition")
- }
- else { // at least one equality expression => generate a join function
- val conditionType = join.getCondition.getType
- val func = getJoinFunction(join, joinInfo)
- val leftKeys = ArrayBuffer.empty[Int]
- val rightKeys = ArrayBuffer.empty[Int]
-
- keyPairs.foreach(pair => {
- leftKeys.add(pair.source)
- rightKeys.add(pair.target)}
- )
-
- new DataSetJoin(
- rel.getCluster,
- traitSet,
- convLeft,
- convRight,
- rel.getRowType,
- join.toString,
- leftKeys.toArray,
- rightKeys.toArray,
- JoinType.INNER,
- null,
- func)
- }
- }
-
- def getJoinFunction(join: FlinkJoin, joinInfo: JoinInfo):
- ((TableConfig, TypeInformation[Any], TypeInformation[Any], TypeInformation[Any]) =>
- FlatJoinFunction[Any, Any, Any]) = {
-
- val func = (
- config: TableConfig,
- leftInputType: TypeInformation[Any],
- rightInputType: TypeInformation[Any],
- returnType: TypeInformation[Any]) => {
-
- val generator = new CodeGenerator(config, leftInputType, Some(rightInputType))
- val conversion = generator.generateConverterResultExpression(
- returnType,
- join.getRowType.getFieldNames)
- var body = ""
-
- if (joinInfo.isEqui) {
- // only equality condition
- body = s"""
- |${conversion.code}
- |${generator.collectorTerm}.collect(${conversion.resultTerm});
- |""".stripMargin
- }
- else {
- val condition = generator.generateExpression(join.getCondition)
- body = s"""
- |${condition.code}
- |if (${condition.resultTerm}) {
- | ${conversion.code}
- | ${generator.collectorTerm}.collect(${conversion.resultTerm});
- |}
- |""".stripMargin
- }
- val genFunction = generator.generateFunction(
- description,
- classOf[FlatJoinFunction[Any, Any, Any]],
- body,
- returnType)
-
- new FlatJoinRunner[Any, Any, Any](
- genFunction.name,
- genFunction.code,
- genFunction.returnType)
- }
- func
- }
-}
-
-object DataSetJoinRule {
- val INSTANCE: RelOptRule = new DataSetJoinRule
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
deleted file mode 100644
index f995201..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.rules.dataset
-
-import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.convert.ConverterRule
-import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetSource}
-import org.apache.flink.api.table.plan.nodes.logical.{FlinkScan, FlinkConvention}
-
-class DataSetScanRule
- extends ConverterRule(
- classOf[FlinkScan],
- FlinkConvention.INSTANCE,
- DataSetConvention.INSTANCE,
- "DataSetScanRule")
-{
-
- def convert(rel: RelNode): RelNode = {
- val scan: FlinkScan = rel.asInstanceOf[FlinkScan]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
-
-
- new DataSetSource(
- rel.getCluster,
- traitSet,
- scan.getTable,
- rel.getRowType
- )
- }
-}
-
-object DataSetScanRule {
- val INSTANCE: RelOptRule = new DataSetScanRule
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetUnionRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetUnionRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetUnionRule.scala
deleted file mode 100644
index a390374..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetUnionRule.scala
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.rules.dataset
-
-import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.convert.ConverterRule
-import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetUnion}
-import org.apache.flink.api.table.plan.nodes.logical.{FlinkUnion, FlinkConvention}
-
-class DataSetUnionRule
- extends ConverterRule(
- classOf[FlinkUnion],
- FlinkConvention.INSTANCE,
- DataSetConvention.INSTANCE,
- "DataSetUnionRule")
-{
-
- def convert(rel: RelNode): RelNode = {
- val union: FlinkUnion = rel.asInstanceOf[FlinkUnion]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
- val convLeft: RelNode = RelOptRule.convert(union.getInput(0), DataSetConvention.INSTANCE)
- val convRight: RelNode = RelOptRule.convert(union.getInput(1), DataSetConvention.INSTANCE)
-
- new DataSetUnion(
- rel.getCluster,
- traitSet,
- convLeft,
- convRight,
- rel.getRowType,
- union.toString)
- }
-}
-
-object DataSetUnionRule {
- val INSTANCE: RelOptRule = new DataSetUnionRule
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala
index a5bfeb6..01a5130 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala
@@ -22,30 +22,32 @@ import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.convert.ConverterRule
import org.apache.calcite.rel.logical.LogicalAggregate
-import org.apache.flink.api.table.plan.nodes.logical.{FlinkAggregate, FlinkConvention}
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetAggregate, DataSetConvention}
+import scala.collection.JavaConversions._
class FlinkAggregateRule
extends ConverterRule(
classOf[LogicalAggregate],
Convention.NONE,
- FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
"FlinkAggregateRule")
{
def convert(rel: RelNode): RelNode = {
val agg: LogicalAggregate = rel.asInstanceOf[LogicalAggregate]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
- val convInput: RelNode = RelOptRule.convert(agg.getInput, FlinkConvention.INSTANCE)
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(agg.getInput, DataSetConvention.INSTANCE)
- new FlinkAggregate(
+ new DataSetAggregate(
rel.getCluster,
traitSet,
convInput,
- agg.indicator,
- agg.getGroupSet,
- agg.getGroupSets,
- agg.getAggCallList)
- }
+ agg.getNamedAggCalls,
+ rel.getRowType,
+ agg.getInput.getRowType,
+ agg.toString,
+ agg.getGroupSet.toArray)
+ }
}
object FlinkAggregateRule {
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
index f40b04d..f5e9c68 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
@@ -22,26 +22,29 @@ import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.convert.ConverterRule
import org.apache.calcite.rel.logical.LogicalCalc
-import org.apache.flink.api.table.plan.nodes.logical.{FlinkCalc, FlinkConvention}
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetCalc, DataSetConvention}
class FlinkCalcRule
extends ConverterRule(
classOf[LogicalCalc],
Convention.NONE,
- FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
"FlinkCalcRule")
{
def convert(rel: RelNode): RelNode = {
val calc: LogicalCalc = rel.asInstanceOf[LogicalCalc]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
- val convInput: RelNode = RelOptRule.convert(calc.getInput, FlinkConvention.INSTANCE)
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(calc.getInput, DataSetConvention.INSTANCE)
- new FlinkCalc(
+ new DataSetCalc(
rel.getCluster,
traitSet,
convInput,
- calc.getProgram)
+ rel.getRowType,
+ calc.getProgram,
+ calc.toString,
+ description)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
index 82f3eaa..c8ce944 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
@@ -24,15 +24,16 @@ import org.apache.calcite.rel.convert.ConverterRule
import org.apache.calcite.rel.logical.LogicalJoin
import org.apache.calcite.rex.{RexInputRef, RexCall}
import org.apache.calcite.sql.fun.SqlStdOperatorTable
-import org.apache.flink.api.table.plan.nodes.logical.{FlinkJoin, FlinkConvention}
-
+import org.apache.flink.api.java.operators.join.JoinType
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetJoin, DataSetConvention}
import scala.collection.JavaConverters._
+import scala.collection.JavaConversions._
class FlinkJoinRule
extends ConverterRule(
classOf[LogicalJoin],
Convention.NONE,
- FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
"FlinkJoinRule")
{
@@ -85,19 +86,27 @@ class FlinkJoinRule
}
def convert(rel: RelNode): RelNode = {
+
val join: LogicalJoin = rel.asInstanceOf[LogicalJoin]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
- val convLeft: RelNode = RelOptRule.convert(join.getInput(0), FlinkConvention.INSTANCE)
- val convRight: RelNode = RelOptRule.convert(join.getInput(1), FlinkConvention.INSTANCE)
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convLeft: RelNode = RelOptRule.convert(join.getInput(0), DataSetConvention.INSTANCE)
+ val convRight: RelNode = RelOptRule.convert(join.getInput(1), DataSetConvention.INSTANCE)
+ val joinInfo = join.analyzeCondition
- new FlinkJoin(
- rel.getCluster,
- traitSet,
- convLeft,
- convRight,
- join.getCondition,
- join.getJoinType,
- join.getVariablesStopped)
+ new DataSetJoin(
+ rel.getCluster,
+ traitSet,
+ convLeft,
+ convRight,
+ rel.getRowType,
+ join.toString,
+ join.getCondition,
+ join.getRowType,
+ joinInfo,
+ joinInfo.pairs.toList,
+ JoinType.INNER,
+ null,
+ description)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinUnionTransposeRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinUnionTransposeRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinUnionTransposeRule.scala
index af54f37..ff3ee8f 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinUnionTransposeRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinUnionTransposeRule.scala
@@ -33,7 +33,7 @@ import org.apache.calcite.rel.core.Union
/**
* This rule is a copy of Calcite's JoinUnionTransposeRule.
* Calcite's implementation checks whether one of the operands is a LogicalUnion,
- * which fails in our case, when it matches with a FlinkUnion.
+ * which fails in our case, when it matches with a DataSetUnion.
* This rule changes this check to match Union, instead of LogicalUnion only.
* The rest of the rule's logic has not been changed.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala
index d789770..21da504 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala
@@ -25,25 +25,24 @@ import org.apache.calcite.rel.core.TableScan
import org.apache.calcite.rel.logical.LogicalTableScan
import org.apache.flink.api.java.DataSet
import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetSource}
-import org.apache.flink.api.table.plan.nodes.logical.{FlinkScan, FlinkConvention}
import org.apache.flink.api.table.plan.schema.DataSetTable
class FlinkScanRule
extends ConverterRule(
classOf[LogicalTableScan],
Convention.NONE,
- FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
"FlinkScanRule")
{
def convert(rel: RelNode): RelNode = {
val scan: TableScan = rel.asInstanceOf[TableScan]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
- val dataSet: DataSet[_] = scan.getTable().unwrap(classOf[DataSetTable[_]]).dataSet
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
- new FlinkScan(
+ new DataSetSource(
rel.getCluster,
traitSet,
- scan.getTable
+ scan.getTable,
+ rel.getRowType
)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22621e02/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
index d9869f8..11600a2 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
@@ -22,30 +22,30 @@ import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.convert.ConverterRule
import org.apache.calcite.rel.logical.LogicalUnion
-import org.apache.flink.api.table.plan.nodes.logical.{FlinkUnion, FlinkConvention}
-
-import scala.collection.JavaConversions._
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetUnion}
class FlinkUnionRule
extends ConverterRule(
classOf[LogicalUnion],
Convention.NONE,
- FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
"FlinkUnionRule")
{
def convert(rel: RelNode): RelNode = {
+
val union: LogicalUnion = rel.asInstanceOf[LogicalUnion]
- val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
- val convInputs = union.getInputs.toList.map(
- RelOptRule.convert(_, FlinkConvention.INSTANCE)
- )
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convLeft: RelNode = RelOptRule.convert(union.getInput(0), DataSetConvention.INSTANCE)
+ val convRight: RelNode = RelOptRule.convert(union.getInput(1), DataSetConvention.INSTANCE)
- new FlinkUnion(
+ new DataSetUnion(
rel.getCluster,
traitSet,
- convInputs,
- union.all)
+ convLeft,
+ convRight,
+ rel.getRowType,
+ union.toString)
}
}