[
https://issues.apache.org/jira/browse/FLINK-3942?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15348797#comment-15348797
]
ASF GitHub Bot commented on FLINK-3942:
---------------------------------------
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/2159#discussion_r68473031
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetIntersect.scala
---
@@ -0,0 +1,153 @@
+/*
+ * 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.{RelOptCost, RelOptPlanner, RelOptCluster,
RelTraitSet}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.metadata.RelMetadataQuery
+import org.apache.calcite.rel.{RelWriter, BiRel, RelNode}
+import org.apache.flink.api.common.functions.FlatJoinFunction
+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.runtime.FlatJoinRunner
+import org.apache.flink.api.table.typeutils.TypeConverter._
+import org.apache.flink.api.table.BatchTableEnvironment
+
+import scala.collection.JavaConverters._
+import scala.collection.JavaConversions._
+
+/**
+ * Flink RelNode which translate Intersect into Join Operator.
+ *
+ */
+class DataSetIntersect(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ left: RelNode,
+ right: RelNode,
+ rowType: RelDataType,
+ all: Boolean,
+ ruleDescription: String)
+ extends BiRel(cluster, traitSet, left, right)
+ with DataSetRel {
+
+ override def deriveRowType() = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs:
java.util.List[RelNode]): RelNode = {
+ new DataSetIntersect(
+ cluster,
+ traitSet,
+ inputs.get(0),
+ inputs.get(1),
+ rowType,
+ all,
+ ruleDescription
+ )
+ }
+
+ override def toString: String = {
+ s"Intersect(intersect: ($intersectSelectionToString))"
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw).item("intersect", intersectSelectionToString)
+ }
+
+ override def computeSelfCost (planner: RelOptPlanner, metadata:
RelMetadataQuery): RelOptCost = {
+ val children = this.getInputs
+ children.foldLeft(planner.getCostFactory.makeCost(0, 0, 0)) { (cost,
child) =>
+ val rowCnt = metadata.getRowCount(child)
+ val rowSize = this.estimateRowSize(child.getRowType)
+ cost.plus(planner.getCostFactory.makeCost(rowCnt, rowCnt, rowCnt *
rowSize))
+ }
+ }
+
+ override def translateToPlan(
+ tableEnv: BatchTableEnvironment,
+ expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
+
+ var leftDataSet: DataSet[Any] = null
+ var rightDataSet: DataSet[Any] = null
+
+ expectedType match {
+ case None =>
+ leftDataSet =
left.asInstanceOf[DataSetRel].translateToPlan(tableEnv)
+ rightDataSet =
+ right.asInstanceOf[DataSetRel].translateToPlan(tableEnv,
Some(leftDataSet.getType))
+ case _ =>
+ leftDataSet =
left.asInstanceOf[DataSetRel].translateToPlan(tableEnv, expectedType)
+ rightDataSet =
right.asInstanceOf[DataSetRel].translateToPlan(tableEnv, expectedType)
+ }
+
+ val config = tableEnv.getConfig
+
+ val returnType = determineReturnType(
+ getRowType,
+ expectedType,
+ config.getNullCheck,
+ config.getEfficientTypeUsage)
+
+ val generator = new CodeGenerator(
+ config,
+ false,
+ leftDataSet.getType,
+ Some(rightDataSet.getType))
+
+ val conversion = generator.generateConverterResultExpression(
+ returnType,
+ left.getRowType.getFieldNames)
+
+
+ val body = s"""
+ |${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)
+
+ val joinOpName = s"intersect: ($intersectSelectionToString)"
+
+ val leftKeys = 0 until left.getRowType.getFieldCount
+ val rightKeys = 0 until left.getRowType.getFieldCount
+
+ val intersectDS = leftDataSet.join(rightDataSet).where(leftKeys:
_*).equalTo(rightKeys: _*)
+ .`with`(joinFun).name(joinOpName)
+
+ if (all) {
+ intersectDS
--- End diff --
An `INTERSECT ALL` cannot be executed as a join because a join builds the
Cartesian product within a join key. However, `INTERSECT ALL` should return as
many identical records as are in both inputs, i.e., with `A = [1, 1, 1, 2, ,2]`
and `B = [1, 2, 2, 3]` `A INTERSECT ALL B` should return `[1, 2, 2]`. Using a
join we'd get `[1, 1, 1, 2, 2, 2, 2]`.
`INTERSECT ALL` can be executed with a `CoGroup` which alternately forwards
both iterators and only emits a record if both iterators had an element. We do
not need a code-generated function for this, as the records are not modified
but simply forwarded.
> Add support for INTERSECT
> -------------------------
>
> Key: FLINK-3942
> URL: https://issues.apache.org/jira/browse/FLINK-3942
> Project: Flink
> Issue Type: New Feature
> Components: Table API & SQL
> Affects Versions: 1.1.0
> Reporter: Fabian Hueske
> Assignee: Jark Wu
> Priority: Minor
>
> Currently, the Table API and SQL do not support INTERSECT.
> INTERSECT can be executed as join on all fields.
> In order to add support for INTERSECT to the Table API and SQL we need to:
> - Implement a {{DataSetIntersect}} class that translates an INTERSECT into a
> DataSet API program using a join on all fields.
> - Implement a {{DataSetIntersectRule}} that translates a Calcite
> {{LogicalIntersect}} into a {{DataSetIntersect}}.
> - Extend the Table API (and validation phase) to provide an intersect()
> method.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
