You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by va...@apache.org on 2016/03/18 14:48:37 UTC
[43/50] [abbrv] flink git commit: [FLINK-3609] [tableAPI] Reorganize
selection of optimization rules
[FLINK-3609] [tableAPI] Reorganize selection of optimization rules
This closes #1797
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e430dee3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e430dee3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e430dee3
Branch: refs/heads/master
Commit: e430dee3529b7424b8c1bf8fccf5f40af6ff540a
Parents: 074ea70
Author: Fabian Hueske <fh...@apache.org>
Authored: Mon Mar 14 14:30:47 2016 +0100
Committer: vasia <va...@apache.org>
Committed: Fri Mar 18 14:44:52 2016 +0100
----------------------------------------------------------------------
.../flink/api/table/codegen/CodeGenerator.scala | 4 +
.../api/table/plan/rules/FlinkRuleSets.scala | 43 +++----
.../FlinkFilterAggregateTransposeRule.scala | 123 +++++++++++++++++++
.../dataSet/FlinkJoinUnionTransposeRule.scala | 110 -----------------
.../table/test/GroupedAggregationsITCase.scala | 17 +++
.../api/scala/table/test/UnionITCase.scala | 16 +++
6 files changed, 179 insertions(+), 134 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e430dee3/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
index de6860f..c7d0ed9 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
@@ -712,6 +712,10 @@ class CodeGenerator(
val operand = operands.head
generateCast(nullCheck, operand, resultType)
+ // as / renaming
+ case AS =>
+ operands.head
+
// string arithmetic
case CONCAT =>
val left = operands.head
http://git-wip-us.apache.org/repos/asf/flink/blob/e430dee3/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 d8cc2b3..94da6f8 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
@@ -29,50 +29,45 @@ object FlinkRuleSets {
*/
val DATASET_OPT_RULES: RuleSet = RuleSets.ofList(
- // filter rules
+ // push a filter into a join
FilterJoinRule.FILTER_ON_JOIN,
+ // push filter into the children of a join
FilterJoinRule.JOIN,
- FilterMergeRule.INSTANCE,
- FilterAggregateTransposeRule.INSTANCE,
+ // push filter through an aggregation
+ FlinkFilterAggregateTransposeRule.INSTANCE,
- // push and merge projection rules
+ // aggregation and projection rules
AggregateProjectMergeRule.INSTANCE,
- ProjectMergeRule.INSTANCE,
+ AggregateProjectPullUpConstantsRule.INSTANCE,
+ // push a projection past a filter or vice versa
ProjectFilterTransposeRule.INSTANCE,
FilterProjectTransposeRule.INSTANCE,
- AggregateProjectPullUpConstantsRule.INSTANCE,
- JoinPushExpressionsRule.INSTANCE,
+ // push a projection to the children of a join
ProjectJoinTransposeRule.INSTANCE,
+ // remove identity project
ProjectRemoveRule.INSTANCE,
+ // reorder sort and projection
SortProjectTransposeRule.INSTANCE,
ProjectSortTransposeRule.INSTANCE,
- // merge and push unions rules
- // TODO: Add a rule to enforce binary unions
+ // join rules
+ JoinPushExpressionsRule.INSTANCE,
+
+ // remove union with only a single child
UnionEliminatorRule.INSTANCE,
- FlinkJoinUnionTransposeRule.LEFT_UNION,
- FlinkJoinUnionTransposeRule.RIGHT_UNION,
- // non-all Union to all-union + distinct
+ // convert non-all union into all-union + distinct
UnionToDistinctRule.INSTANCE,
- // aggregation rules
+ // remove aggregation if it does not aggregate and input is already distinct
AggregateRemoveRule.INSTANCE,
+ // push aggregate through join
AggregateJoinTransposeRule.EXTENDED,
+ // aggregate union rule
AggregateUnionAggregateRule.INSTANCE,
- // deactivate this rule temporarily
- // AggregateReduceFunctionsRule.INSTANCE,
- AggregateExpandDistinctAggregatesRule.INSTANCE,
// remove unnecessary sort rule
SortRemoveRule.INSTANCE,
- // join reordering rules
- JoinPushThroughJoinRule.LEFT,
- JoinPushThroughJoinRule.RIGHT,
- JoinAssociateRule.INSTANCE,
- JoinCommuteRule.INSTANCE,
- JoinCommuteRule.SWAP_OUTER,
-
// simplify expressions rules
ReduceExpressionsRule.CALC_INSTANCE,
ReduceExpressionsRule.JOIN_INSTANCE,
@@ -93,7 +88,7 @@ object FlinkRuleSets {
ProjectToCalcRule.INSTANCE,
CalcMergeRule.INSTANCE,
- // translate to logical Flink nodes
+ // translate to Flink DataSet nodes
DataSetAggregateRule.INSTANCE,
DataSetCalcRule.INSTANCE,
DataSetJoinRule.INSTANCE,
http://git-wip-us.apache.org/repos/asf/flink/blob/e430dee3/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/FlinkFilterAggregateTransposeRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/FlinkFilterAggregateTransposeRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/FlinkFilterAggregateTransposeRule.scala
new file mode 100644
index 0000000..c27fc75
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/FlinkFilterAggregateTransposeRule.scala
@@ -0,0 +1,123 @@
+/*
+ * 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
+import org.apache.calcite.plan.RelOptRuleCall
+import org.apache.calcite.plan.RelOptUtil
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.Aggregate
+import org.apache.calcite.rel.core.Filter
+import org.apache.calcite.rel.core.RelFactories
+import org.apache.calcite.rex.RexNode
+import org.apache.calcite.tools.RelBuilder
+import org.apache.calcite.tools.RelBuilderFactory
+import org.apache.calcite.util.ImmutableBitSet
+import com.google.common.collect.ImmutableList
+
+import scala.collection.JavaConversions._
+
+/**
+ * This rule is a (fixed) copy of the FilterAggregateTransposeRule of Apache Calcite.
+ *
+ * A fix for this rule is contained Calcite's master branch.
+ * This custom rule can be removed once Calcite 1.7 is released and our dependency adjusted.
+ *
+ * Planner rule that pushes a `org.apache.calcite.rel.core.Filter`
+ * past a `org.apache.calcite.rel.core.Aggregate`.
+ *
+ * @see org.apache.calcite.rel.rules.AggregateFilterTransposeRule
+ */
+class FlinkFilterAggregateTransposeRule(
+ filterClass: Class[_ <: Filter],
+ builderFactory: RelBuilderFactory,
+ aggregateClass: Class[_ <: Aggregate])
+ extends RelOptRule(
+ RelOptRule.operand(filterClass, RelOptRule.operand(aggregateClass, RelOptRule.any)),
+ builderFactory,
+ null)
+{
+
+ def onMatch(call: RelOptRuleCall) {
+ val filterRel: Filter = call.rel(0)
+ val aggRel: Aggregate = call.rel(1)
+ val conditions = RelOptUtil.conjunctions(filterRel.getCondition).toList
+ val rexBuilder = filterRel.getCluster.getRexBuilder
+ val origFields = aggRel.getRowType.getFieldList.toList
+
+ // Fixed computation of adjustments
+ val adjustments = aggRel.getGroupSet.asList().zipWithIndex.map {
+ case (g, i) => g - i
+ }.toArray
+
+ var pushedConditions: List[RexNode] = Nil
+ var remainingConditions: List[RexNode] = Nil
+
+ for (condition <- conditions) {
+ val rCols: ImmutableBitSet = RelOptUtil.InputFinder.bits(condition)
+ if (canPush(aggRel, rCols)) {
+ pushedConditions = condition.accept(
+ new RelOptUtil.RexInputConverter(
+ rexBuilder,
+ origFields,
+ aggRel.getInput(0).getRowType.getFieldList,
+ adjustments)) :: pushedConditions
+ }
+ else {
+ remainingConditions = condition :: remainingConditions
+ }
+ }
+ val builder: RelBuilder = call.builder
+ var rel: RelNode = builder.push(aggRel.getInput).filter(pushedConditions).build
+ if (rel eq aggRel.getInput(0)) {
+ return
+ }
+ rel = aggRel.copy(aggRel.getTraitSet, ImmutableList.of(rel))
+ rel = builder.push(rel).filter(remainingConditions).build
+ call.transformTo(rel)
+ }
+
+ private def canPush(aggregate: Aggregate, rCols: ImmutableBitSet): Boolean = {
+ val groupKeys: ImmutableBitSet = ImmutableBitSet.range(0, aggregate.getGroupSet.cardinality)
+ if (!groupKeys.contains(rCols)) {
+ return false
+ }
+ if (aggregate.indicator) {
+ import scala.collection.JavaConversions._
+ for (groupingSet <- aggregate.getGroupSets) {
+ if (!groupingSet.contains(rCols)) {
+ return false
+ }
+ }
+ }
+ true
+ }
+}
+
+object FlinkFilterAggregateTransposeRule {
+ /** The default instance of `FilterAggregateTransposeRule`.
+ *
+ * It matches any kind of agg. or filter
+ */
+ val INSTANCE: FlinkFilterAggregateTransposeRule =
+ new FlinkFilterAggregateTransposeRule(
+ classOf[Filter],
+ RelFactories.LOGICAL_BUILDER,
+ classOf[Aggregate])
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/e430dee3/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
deleted file mode 100644
index 32b53e2..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/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.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/e430dee3/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala
index 8b9609f..cbed9e3 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala
@@ -174,4 +174,21 @@ class GroupedAggregationsITCase(mode: TestExecutionMode) extends MultipleProgram
val results = t.toDataSet[Row].collect()
TestBaseUtils.compareResultAsText(results.asJava, expected)
}
+
+ @Test
+ def testGroupedAggregateWithFilter(): 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)
+ .where('b === 2)
+
+ val expected = "2,5\n"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e430dee3/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala
index e3d1b18..1263eba 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala
@@ -51,6 +51,22 @@ class UnionITCase(
}
@Test
+ def testTernaryUnion(): Unit = {
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
+ val ds2 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
+ val ds3 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
+
+ val unionDs = ds1.unionAll(ds2).unionAll(ds3).select('c)
+
+ val results = unionDs.toDataSet[Row].collect()
+ val expected = "Hi\n" + "Hello\n" + "Hello world\n" +
+ "Hi\n" + "Hello\n" + "Hello world\n" +
+ "Hi\n" + "Hello\n" + "Hello world\n"
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
+ }
+
+ @Test
def testUnionWithFilter(): Unit = {
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)