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/01/23 10:55:00 UTC
[4/4] flink git commit: [FLINK-3223] Translate Table API calls to
Calcite RelNodes.
[FLINK-3223] Translate Table API calls to Calcite RelNodes.
This is an intermediate step to port the Table API on top of Calcite (FLINK-3221).
This commit:
- Adds Calcite as dependency to flink-table.
- Translates Table API calls directly into Calcite RelNodes.
- Modifies tests to check only the translation into logical plans but not the execution of Table API queries.
- Deactivates a few tests that are not supported yet.
- Removes a lot of the former Table API translation code.
- Removes bitwise operators from the Table API.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e4ee0c6d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e4ee0c6d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e4ee0c6d
Branch: refs/heads/tableOnCalcite
Commit: e4ee0c6d7393b1d555b08c8f515cf9724d25c7fb
Parents: a65cd8d
Author: Fabian Hueske <fh...@apache.org>
Authored: Fri Jan 15 17:46:39 2016 +0100
Committer: Fabian Hueske <fh...@apache.org>
Committed: Sat Jan 23 10:45:58 2016 +0100
----------------------------------------------------------------------
flink-libraries/flink-table/pom.xml | 6 +
.../api/java/table/JavaBatchTranslator.scala | 339 ++-----------------
.../java/table/JavaStreamingTranslator.scala | 241 -------------
.../flink/api/java/table/TableEnvironment.scala | 43 +--
.../api/scala/table/DataStreamConversions.scala | 68 ----
.../api/scala/table/ScalaBatchTranslator.scala | 26 +-
.../scala/table/ScalaStreamingTranslator.scala | 58 ----
.../api/scala/table/TableConversions.scala | 12 +-
.../flink/api/scala/table/expressionDsl.scala | 5 -
.../apache/flink/api/scala/table/package.scala | 21 +-
.../org/apache/flink/api/table/Table.scala | 236 +++++++++----
.../table/codegen/ExpressionCodeGenerator.scala | 35 --
.../api/table/expressions/aggregations.scala | 40 +--
.../analysis/ExtractEquiJoinFields.scala | 70 ----
.../expressions/analysis/GroupByAnalyzer.scala | 51 ---
.../expressions/analysis/InsertAutoCasts.scala | 92 -----
.../analysis/PredicateAnalyzer.scala | 35 --
.../analysis/ResolveFieldReferences.scala | 60 ----
.../analysis/SelectionAnalyzer.scala | 36 --
.../table/expressions/analysis/TypeCheck.scala | 57 ----
.../expressions/analysis/VerifyBoolean.scala | 41 ---
.../analysis/VerifyNoAggregates.scala | 53 ---
.../analysis/VerifyNoNestedAggregates.scala | 54 ---
.../api/table/expressions/arithmetic.scala | 56 +--
.../flink/api/table/expressions/cast.scala | 5 +
.../api/table/expressions/fieldExpression.scala | 5 +
.../api/table/parser/ExpressionParser.scala | 16 +-
.../api/table/plan/ExpandAggregations.scala | 147 --------
.../flink/api/table/plan/PlanTranslator.scala | 133 ++------
.../api/table/plan/RexNodeTranslator.scala | 184 ++++++++++
.../flink/api/table/plan/TypeConverter.scala | 54 +++
.../flink/api/table/plan/operations.scala | 134 --------
.../api/table/plan/operators/DataSetTable.scala | 66 ++++
.../apache/flink/api/table/plan/package.scala | 24 --
.../apache/flink/api/table/trees/Analyzer.scala | 43 ---
.../org/apache/flink/api/table/trees/Rule.scala | 30 --
.../examples/scala/StreamingTableFilter.scala | 90 -----
.../api/java/table/test/AggregationsITCase.java | 67 ++--
.../flink/api/java/table/test/AsITCase.java | 68 ++--
.../api/java/table/test/CastingITCase.java | 62 ++--
.../api/java/table/test/ExpressionsITCase.java | 83 +----
.../flink/api/java/table/test/FilterITCase.java | 58 ++--
.../table/test/GroupedAggregationsITCase.java | 35 +-
.../flink/api/java/table/test/JoinITCase.java | 70 ++--
.../api/java/table/test/PojoGroupingITCase.java | 18 +-
.../flink/api/java/table/test/SelectITCase.java | 70 ++--
.../api/java/table/test/SqlExplainITCase.java | 7 +
.../table/test/StringExpressionsITCase.java | 45 +--
.../flink/api/java/table/test/UnionITCase.java | 44 +--
.../scala/table/test/PageRankTableITCase.java | 7 +-
.../scala/table/test/TypeExceptionTest.scala | 42 ---
.../scala/table/test/AggregationsITCase.scala | 104 +++---
.../flink/api/scala/table/test/AsITCase.scala | 74 ++--
.../api/scala/table/test/CastingITCase.scala | 57 ++--
.../scala/table/test/ExpressionsITCase.scala | 90 ++---
.../api/scala/table/test/FilterITCase.scala | 59 ++--
.../table/test/GroupedAggreagationsITCase.scala | 115 -------
.../table/test/GroupedAggregationsITCase.scala | 97 ++++++
.../flink/api/scala/table/test/JoinITCase.scala | 73 ++--
.../api/scala/table/test/SelectITCase.scala | 99 +++---
.../api/scala/table/test/SqlExplainITCase.scala | 198 +++++------
.../table/test/StringExpressionsITCase.scala | 52 +--
.../api/scala/table/test/UnionITCase.scala | 34 +-
63 files changed, 1389 insertions(+), 3005 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/pom.xml
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/pom.xml b/flink-libraries/flink-table/pom.xml
index 19951cb..818b537 100644
--- a/flink-libraries/flink-table/pom.xml
+++ b/flink-libraries/flink-table/pom.xml
@@ -100,6 +100,12 @@ under the License.
<version>${jackson.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.calcite</groupId>
+ <artifactId>calcite-core</artifactId>
+ <version>1.5.0</version>
+ </dependency>
+
</dependencies>
<build>
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/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 9dc9297..f3f4e9d 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
@@ -18,22 +18,14 @@
package org.apache.flink.api.java.table
-import java.lang.reflect.Modifier
-
-import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint
+import org.apache.calcite.plan.RelOptUtil
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.tools.{RelBuilder, Frameworks}
import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.java.aggregation.AggregationFunction
-import org.apache.flink.api.java.operators.JoinOperator.EquiJoin
-import org.apache.flink.api.java.operators.Keys.ExpressionKeys
-import org.apache.flink.api.java.operators.{GroupReduceOperator, Keys, MapOperator, UnsortedGrouping}
import org.apache.flink.api.java.{DataSet => JavaDataSet}
-import org.apache.flink.api.table.expressions.analysis.ExtractEquiJoinFields
import org.apache.flink.api.table.plan._
-import org.apache.flink.api.table.runtime._
-import org.apache.flink.api.table.expressions._
-import org.apache.flink.api.table.typeinfo.{RenameOperator, RenamingProxyTypeInfo, RowTypeInfo}
-import org.apache.flink.api.table.{ExpressionException, Row, Table}
+import org.apache.flink.api.table.plan.operators.DataSetTable
+import org.apache.flink.api.table.Table
/**
* [[PlanTranslator]] for creating [[Table]]s from Java [[org.apache.flink.api.java.DataSet]]s and
@@ -45,302 +37,41 @@ class JavaBatchTranslator extends PlanTranslator {
override def createTable[A](
repr: Representation[A],
- inputType: CompositeType[A],
- expressions: Array[Expression],
- resultFields: Seq[(String, TypeInformation[_])]): Table = {
-
- val rowDataSet = createSelect(expressions, repr, inputType)
-
- Table(Root(rowDataSet, resultFields))
+ fieldNames: Array[String]): Table = {
+
+ // create table representation from DataSet
+ val dataSetTable = new DataSetTable[A](
+ repr.asInstanceOf[JavaDataSet[A]],
+ fieldNames
+ )
+
+ // register table in Cascading schema
+ val schema = Frameworks.createRootSchema(true)
+ val tableName = repr.hashCode().toString
+ schema.add(tableName, dataSetTable)
+
+ // initialize RelBuilder
+ val frameworkConfig = Frameworks
+ .newConfigBuilder
+ .defaultSchema(schema)
+ .build
+ val relBuilder = RelBuilder.create(frameworkConfig)
+
+ // create table scan operator
+ relBuilder.scan(tableName)
+ new Table(relBuilder.build(), relBuilder)
}
- override def translate[A](op: PlanNode)(implicit tpe: TypeInformation[A]): JavaDataSet[A] = {
-
- if (tpe.getTypeClass == classOf[Row]) {
- // shortcut for DataSet[Row]
- return translateInternal(op).asInstanceOf[JavaDataSet[A]]
- }
-
- val clazz = tpe.getTypeClass
- if (clazz.isMemberClass && !Modifier.isStatic(clazz.getModifiers)) {
- throw new ExpressionException("Cannot create DataSet of type " +
- clazz.getName + ". Only top-level classes or static member classes are supported.")
- }
-
-
- if (!implicitly[TypeInformation[A]].isInstanceOf[CompositeType[A]]) {
- throw new ExpressionException(
- "A Table can only be converted to composite types, type is: " +
- implicitly[TypeInformation[A]] +
- ". Composite types would be tuples, case classes and POJOs.")
- }
-
- val resultSet = translateInternal(op)
-
- val resultType = resultSet.getType.asInstanceOf[RowTypeInfo]
+ override def translate[A](lPlan: RelNode)(implicit tpe: TypeInformation[A]): JavaDataSet[A] = {
- val outputType = implicitly[TypeInformation[A]].asInstanceOf[CompositeType[A]]
+ println(RelOptUtil.toString(lPlan))
- val resultNames = resultType.getFieldNames
- val outputNames = outputType.getFieldNames.toSeq
+ // TODO: optimize & translate:
+ // - optimize RelNode plan
+ // - translate to Flink RelNode plan
+ // - generate DataSet program
- if (resultNames.toSet != outputNames.toSet) {
- throw new ExpressionException(s"Expression result type $resultType does not have the same " +
- s"fields as output type $outputType")
- }
-
- for (f <- outputNames) {
- val in = resultType.getTypeAt(resultType.getFieldIndex(f))
- val out = outputType.getTypeAt(outputType.getFieldIndex(f))
- if (!in.equals(out)) {
- throw new ExpressionException(s"Types for field $f differ on input $resultType and " +
- s"output $outputType.")
- }
- }
-
- val outputFields = outputNames map {
- f => ResolvedFieldReference(f, resultType.getTypeAt(f))
- }
-
- val function = new ExpressionSelectFunction(
- resultSet.getType.asInstanceOf[RowTypeInfo],
- outputType,
- outputFields)
-
- val opName = s"select(${outputFields.mkString(",")})"
- val operator = new MapOperator(resultSet, outputType, function, opName)
-
- operator
+ null
}
- private def translateInternal(op: PlanNode): JavaDataSet[Row] = {
- op match {
- case Root(dataSet: JavaDataSet[Row], resultFields) =>
- dataSet
-
- case Root(_, _) =>
- throw new ExpressionException("Invalid Root for JavaBatchTranslator: " + op + ". " +
- "Did you try converting a Table based on a DataSet to a DataStream or vice-versa?")
-
- case GroupBy(_, fields) =>
- throw new ExpressionException("Dangling GroupBy operation. Did you forget a " +
- "SELECT statement?")
-
- case As(input, newNames) =>
- val translatedInput = translateInternal(input)
- val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
- val proxyType = new RenamingProxyTypeInfo[Row](inType, newNames.toArray)
- new RenameOperator(translatedInput, proxyType)
-
- case sel@Select(Filter(Join(leftInput, rightInput), predicate), selection) =>
-
- val expandedInput = ExpandAggregations(sel)
-
- if (expandedInput.eq(sel)) {
- val translatedLeftInput = translateInternal(leftInput)
- val translatedRightInput = translateInternal(rightInput)
- val leftInType = translatedLeftInput.getType.asInstanceOf[CompositeType[Row]]
- val rightInType = translatedRightInput.getType.asInstanceOf[CompositeType[Row]]
-
- createJoin(
- predicate,
- selection,
- translatedLeftInput,
- translatedRightInput,
- leftInType,
- rightInType,
- JoinHint.OPTIMIZER_CHOOSES)
- } else {
- translateInternal(expandedInput)
- }
-
- case Filter(Join(leftInput, rightInput), predicate) =>
- val translatedLeftInput = translateInternal(leftInput)
- val translatedRightInput = translateInternal(rightInput)
- val leftInType = translatedLeftInput.getType.asInstanceOf[CompositeType[Row]]
- val rightInType = translatedRightInput.getType.asInstanceOf[CompositeType[Row]]
-
- createJoin(
- predicate,
- leftInput.outputFields.map( f => ResolvedFieldReference(f._1, f._2)) ++
- rightInput.outputFields.map( f => ResolvedFieldReference(f._1, f._2)),
- translatedLeftInput,
- translatedRightInput,
- leftInType,
- rightInType,
- JoinHint.OPTIMIZER_CHOOSES)
-
- case Join(leftInput, rightInput) =>
- throw new ExpressionException("Join without filter condition encountered. " +
- "Did you forget to add .where(...) ?")
-
- case sel@Select(input, selection) =>
-
- val expandedInput = ExpandAggregations(sel)
-
- if (expandedInput.eq(sel)) {
- val translatedInput = input match {
- case GroupBy(groupByInput, groupExpressions) =>
- val translatedGroupByInput = translateInternal(groupByInput)
- val inType = translatedGroupByInput.getType.asInstanceOf[CompositeType[Row]]
-
- val keyIndices = groupExpressions map {
- case fe: ResolvedFieldReference => inType.getFieldIndex(fe.name)
- case e =>
- throw new ExpressionException(s"Expression $e is not a valid key expression.")
- }
-
- val keys = new Keys.ExpressionKeys(keyIndices.toArray, inType, false)
- val grouping = new UnsortedGrouping(translatedGroupByInput, keys)
-
- new GroupReduceOperator(
- grouping,
- inType,
- new NoExpressionAggregateFunction(),
- "Nop Expression Aggregation")
-
- case _ => translateInternal(input)
- }
-
- val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
- val inputFields = inType.getFieldNames
- createSelect(
- selection,
- translatedInput,
- inType)
- } else {
- translateInternal(expandedInput)
- }
-
- case agg@Aggregate(GroupBy(input, groupExpressions), aggregations) =>
- val translatedInput = translateInternal(input)
- val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
-
- val keyIndices = groupExpressions map {
- case fe: ResolvedFieldReference => inType.getFieldIndex(fe.name)
- case e => throw new ExpressionException(s"Expression $e is not a valid key expression.")
- }
-
- val keys = new Keys.ExpressionKeys(keyIndices.toArray, inType, false)
-
- val grouping = new UnsortedGrouping(translatedInput, keys)
-
- val aggFunctions: Seq[AggregationFunction[Any]] = aggregations map {
- case (fieldName, fun) =>
- fun.getFactory.createAggregationFunction[Any](
- inType.getTypeAt[Any](inType.getFieldIndex(fieldName)).getTypeClass)
- }
-
- val aggIndices = aggregations map {
- case (fieldName, _) =>
- inType.getFieldIndex(fieldName)
- }
-
- val result = new GroupReduceOperator(
- grouping,
- inType,
- new ExpressionAggregateFunction(aggIndices, aggFunctions),
- "Expression Aggregation: " + agg)
-
- result
-
- case agg@Aggregate(input, aggregations) =>
- val translatedInput = translateInternal(input)
- val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
-
- val aggFunctions: Seq[AggregationFunction[Any]] = aggregations map {
- case (fieldName, fun) =>
- fun.getFactory.createAggregationFunction[Any](
- inType.getTypeAt[Any](inType.getFieldIndex(fieldName)).getTypeClass)
- }
-
- val aggIndices = aggregations map {
- case (fieldName, _) =>
- inType.getFieldIndex(fieldName)
- }
-
- val result = new GroupReduceOperator(
- translatedInput,
- inType,
- new ExpressionAggregateFunction(aggIndices, aggFunctions),
- "Expression Aggregation: " + agg)
-
- result
-
-
- case Filter(input, predicate) =>
- val translatedInput = translateInternal(input)
- val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
- val filter = new ExpressionFilterFunction[Row](predicate, inType)
- translatedInput.filter(filter).name(predicate.toString)
-
- case uni@UnionAll(left, right) =>
- val translatedLeft = translateInternal(left)
- val translatedRight = translateInternal(right)
- translatedLeft.union(translatedRight).name("Union: " + uni)
- }
- }
-
- private def createSelect[I](
- fields: Seq[Expression],
- input: JavaDataSet[I],
- inputType: CompositeType[I]): JavaDataSet[Row] = {
-
- fields foreach {
- f =>
- if (f.exists(_.isInstanceOf[Aggregation])) {
- throw new ExpressionException("Found aggregate in " + fields.mkString(", ") + ".")
- }
-
- }
-
- val resultType = new RowTypeInfo(fields)
-
- val function = new ExpressionSelectFunction(inputType, resultType, fields)
-
- val opName = s"select(${fields.mkString(",")})"
- val operator = new MapOperator(input, resultType, function, opName)
-
- operator
- }
-
- private def createJoin[L, R](
- predicate: Expression,
- fields: Seq[Expression],
- leftInput: JavaDataSet[L],
- rightInput: JavaDataSet[R],
- leftType: CompositeType[L],
- rightType: CompositeType[R],
- joinHint: JoinHint): JavaDataSet[Row] = {
-
- val resultType = new RowTypeInfo(fields)
-
- val (reducedPredicate, leftFields, rightFields) =
- ExtractEquiJoinFields(leftType, rightType, predicate)
-
- if (leftFields.isEmpty || rightFields.isEmpty) {
- throw new ExpressionException("Could not derive equi-join predicates " +
- "for predicate " + predicate + ".")
- }
-
- val leftKey = new ExpressionKeys[L](leftFields, leftType)
- val rightKey = new ExpressionKeys[R](rightFields, rightType)
-
- val joiner = new ExpressionJoinFunction[L, R, Row](
- reducedPredicate,
- leftType,
- rightType,
- resultType,
- fields)
-
- new EquiJoin[L, R, Row](
- leftInput,
- rightInput,
- leftKey,
- rightKey,
- joiner,
- resultType,
- joinHint,
- predicate.toString)
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaStreamingTranslator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaStreamingTranslator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaStreamingTranslator.scala
deleted file mode 100644
index a37c892..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaStreamingTranslator.scala
+++ /dev/null
@@ -1,241 +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.java.table
-
-import java.lang.reflect.Modifier
-import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.table.plan._
-import org.apache.flink.api.table.runtime.{ExpressionFilterFunction, ExpressionSelectFunction}
-import org.apache.flink.api.table.expressions._
-import org.apache.flink.api.table.typeinfo.RowTypeInfo
-import org.apache.flink.api.table.{ExpressionException, Row, Table}
-import org.apache.flink.streaming.api.datastream.DataStream
-import org.apache.flink.streaming.api.operators.StreamMap
-
-/**
- * [[PlanTranslator]] for creating [[Table]]s from Java [[DataStream]]s and
- * translating them back to Java [[DataStream]]s.
- *
- * This is very limited right now. Only select and filter are implemented. Also, the expression
- * operations must be extended to allow windowing operations.
- */
-
-class JavaStreamingTranslator extends PlanTranslator {
-
- type Representation[A] = DataStream[A]
-
- override def createTable[A](
- repr: Representation[A],
- inputType: CompositeType[A],
- expressions: Array[Expression],
- resultFields: Seq[(String, TypeInformation[_])]): Table = {
-
- val rowDataStream = createSelect(expressions, repr, inputType)
-
- new Table(Root(rowDataStream, resultFields))
- }
-
- override def translate[A](op: PlanNode)(implicit tpe: TypeInformation[A]): DataStream[A] = {
-
- if (tpe.getTypeClass == classOf[Row]) {
- // shortcut for DataSet[Row]
- return translateInternal(op).asInstanceOf[DataStream[A]]
- }
-
- val clazz = tpe.getTypeClass
- if (clazz.isMemberClass && !Modifier.isStatic(clazz.getModifiers)) {
- throw new ExpressionException("Cannot create DataStream of type " +
- clazz.getName + ". Only top-level classes or static member classes are supported.")
- }
-
- if (!implicitly[TypeInformation[A]].isInstanceOf[CompositeType[A]]) {
- throw new ExpressionException(
- "A Table can only be converted to composite types, type is: " +
- implicitly[TypeInformation[A]] +
- ". Composite types would be tuples, case classes and POJOs.")
-
- }
-
- val resultSet = translateInternal(op)
-
- val resultType = resultSet.getType.asInstanceOf[RowTypeInfo]
-
- val outputType = implicitly[TypeInformation[A]].asInstanceOf[CompositeType[A]]
-
- val resultNames = resultType.getFieldNames
- val outputNames = outputType.getFieldNames.toSeq
-
- if (resultNames.toSet != outputNames.toSet) {
- throw new ExpressionException(s"Expression result type $resultType does not have the same" +
- s"fields as output type $outputType")
- }
-
- for (f <- outputNames) {
- val in = resultType.getTypeAt(resultType.getFieldIndex(f))
- val out = outputType.getTypeAt(outputType.getFieldIndex(f))
- if (!in.equals(out)) {
- throw new ExpressionException(s"Types for field $f differ on input $resultType and " +
- s"output $outputType.")
- }
- }
-
- val outputFields = outputNames map {
- f => ResolvedFieldReference(f, resultType.getTypeAt(f))
- }
-
- val function = new ExpressionSelectFunction(
- resultSet.getType.asInstanceOf[RowTypeInfo],
- outputType,
- outputFields)
-
- val opName = s"select(${outputFields.mkString(",")})"
-
- resultSet.transform(opName, outputType, new StreamMap[Row, A](function))
- }
-
- private def translateInternal(op: PlanNode): DataStream[Row] = {
- op match {
- case Root(dataSet: DataStream[Row], resultFields) =>
- dataSet
-
- case Root(_, _) =>
- throw new ExpressionException("Invalid Root for JavaStreamingTranslator: " + op + ". " +
- "Did you try converting a Table based on a DataSet to a DataStream or vice-versa?")
-
- case GroupBy(_, fields) =>
- throw new ExpressionException("Dangling GroupBy operation. Did you forget a " +
- "SELECT statement?")
-
- case As(input, newNames) =>
- throw new ExpressionException("As operation for Streams not yet implemented.")
-
- case sel@Select(Filter(Join(leftInput, rightInput), predicate), selection) =>
-
- val expandedInput = ExpandAggregations(sel)
-
- if (expandedInput.eq(sel)) {
- val translatedLeftInput = translateInternal(leftInput)
- val translatedRightInput = translateInternal(rightInput)
- val leftInType = translatedLeftInput.getType.asInstanceOf[CompositeType[Row]]
- val rightInType = translatedRightInput.getType.asInstanceOf[CompositeType[Row]]
-
- createJoin(
- predicate,
- selection,
- translatedLeftInput,
- translatedRightInput,
- leftInType,
- rightInType,
- JoinHint.OPTIMIZER_CHOOSES)
- } else {
- translateInternal(expandedInput)
- }
-
- case Filter(Join(leftInput, rightInput), predicate) =>
- val translatedLeftInput = translateInternal(leftInput)
- val translatedRightInput = translateInternal(rightInput)
- val leftInType = translatedLeftInput.getType.asInstanceOf[CompositeType[Row]]
- val rightInType = translatedRightInput.getType.asInstanceOf[CompositeType[Row]]
-
- createJoin(
- predicate,
- leftInput.outputFields.map( f => ResolvedFieldReference(f._1, f._2)) ++
- rightInput.outputFields.map( f => ResolvedFieldReference(f._1, f._2)),
- translatedLeftInput,
- translatedRightInput,
- leftInType,
- rightInType,
- JoinHint.OPTIMIZER_CHOOSES)
-
- case Join(leftInput, rightInput) =>
- throw new ExpressionException("Join without filter condition encountered. " +
- "Did you forget to add .where(...) ?")
-
- case sel@Select(input, selection) =>
-
- val expandedInput = ExpandAggregations(sel)
-
- if (expandedInput.eq(sel)) {
- // no expansions took place
- val translatedInput = translateInternal(input)
- val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
- val inputFields = inType.getFieldNames
- createSelect(
- selection,
- translatedInput,
- inType)
- } else {
- translateInternal(expandedInput)
- }
-
- case agg@Aggregate(GroupBy(input, groupExpressions), aggregations) =>
- throw new ExpressionException("Aggregate operation for Streams not yet implemented.")
-
- case agg@Aggregate(input, aggregations) =>
- throw new ExpressionException("Aggregate operation for Streams not yet implemented.")
-
- case Filter(input, predicate) =>
- val translatedInput = translateInternal(input)
- val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
- val filter = new ExpressionFilterFunction[Row](predicate, inType)
- translatedInput.filter(filter)
-
- case UnionAll(left, right) =>
- val translatedLeft = translateInternal(left)
- val translatedRight = translateInternal(right)
- translatedLeft.union(translatedRight)
- }
- }
-
- private def createSelect[I](
- fields: Seq[Expression],
- input: DataStream[I],
- inputType: CompositeType[I]): DataStream[Row] = {
-
- fields foreach {
- f =>
- if (f.exists(_.isInstanceOf[Aggregation])) {
- throw new ExpressionException("Found aggregate in " + fields.mkString(", ") + ".")
- }
-
- }
-
- val resultType = new RowTypeInfo(fields)
-
- val function = new ExpressionSelectFunction(inputType, resultType, fields)
-
- val opName = s"select(${fields.mkString(",")})"
-
- input.transform(opName, resultType, new StreamMap[I, Row](function))
- }
-
- private def createJoin[L, R](
- predicate: Expression,
- fields: Seq[Expression],
- leftInput: DataStream[L],
- rightInput: DataStream[R],
- leftType: CompositeType[L],
- rightType: CompositeType[R],
- joinHint: JoinHint): DataStream[Row] = {
-
- throw new ExpressionException("Join operation for Streams not yet implemented.")
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala
index 5614031..01e38db 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala
@@ -21,12 +21,11 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.DataSet
import org.apache.flink.api.java.typeutils.TypeExtractor
import org.apache.flink.api.table.Table
-import org.apache.flink.streaming.api.datastream.DataStream
/**
* Environment for working with the Table API.
*
- * This can be used to convert [[DataSet]] or [[DataStream]] to a [[Table]] and back again. You
+ * This can be used to convert a [[DataSet]] to a [[Table]] and back again. You
* can also use the provided methods to create a [[Table]] directly from a data source.
*/
class TableEnvironment {
@@ -58,32 +57,6 @@ class TableEnvironment {
}
/**
- * Transforms the given DataStream to a [[org.apache.flink.api.table.Table]].
- * The fields of the DataStream type are renamed to the given set of fields:
- *
- * Example:
- *
- * {{{
- * tableEnv.fromDataStream(set, "a, b")
- * }}}
- *
- * This will transform the set containing elements of two fields to a table where the fields
- * are named a and b.
- */
- def fromDataStream[T](set: DataStream[T], fields: String): Table = {
- new JavaStreamingTranslator().createTable(set, fields)
- }
-
- /**
- * Transforms the given DataStream to a [[org.apache.flink.api.table.Table]].
- * The fields of the DataStream type are used to name the
- * [[org.apache.flink.api.table.Table]] fields.
- */
- def fromDataStream[T](set: DataStream[T]): Table = {
- new JavaStreamingTranslator().createTable(set)
- }
-
- /**
* Converts the given [[org.apache.flink.api.table.Table]] to
* a DataSet. The given type must have exactly the same fields as the
* [[org.apache.flink.api.table.Table]]. That is, the names of the
@@ -91,21 +64,9 @@ class TableEnvironment {
*/
@SuppressWarnings(Array("unchecked"))
def toDataSet[T](table: Table, clazz: Class[T]): DataSet[T] = {
- new JavaBatchTranslator().translate[T](table.operation)(
+ new JavaBatchTranslator().translate[T](table.relNode)(
TypeExtractor.createTypeInfo(clazz).asInstanceOf[TypeInformation[T]])
}
- /**
- * Converts the given [[org.apache.flink.api.table.Table]] to
- * a DataStream. The given type must have exactly the same fields as the
- * [[org.apache.flink.api.table.Table]]. That is, the names of the
- * fields and the types must match.
- */
- @SuppressWarnings(Array("unchecked"))
- def toDataStream[T](table: Table, clazz: Class[T]): DataStream[T] = {
- new JavaStreamingTranslator().translate[T](table.operation)(
- TypeExtractor.createTypeInfo(clazz).asInstanceOf[TypeInformation[T]])
-
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataStreamConversions.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataStreamConversions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataStreamConversions.scala
deleted file mode 100644
index 47bd100..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataStreamConversions.scala
+++ /dev/null
@@ -1,68 +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
-
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.table._
-import org.apache.flink.api.table.expressions.{Expression, UnresolvedFieldReference}
-import org.apache.flink.streaming.api.scala.DataStream
-
-class DataStreamConversions[T](stream: DataStream[T], inputType: CompositeType[T]) {
-
- /**
- * Converts the [[DataStream]] to a [[Table]]. The field names can be specified like this:
- *
- * {{{
- * val in: DataSet[(String, Int)] = ...
- * val table = in.as('a, 'b)
- * }}}
- *
- * This results in a [[Table]] that has field `a` of type `String` and field `b`
- * of type `Int`.
- */
-
- def as(fields: Expression*): Table = {
- new ScalaStreamingTranslator().createTable(
- stream,
- fields.toArray,
- checkDeterministicFields = true)
- }
-
- /**
- * Converts the [[DataStream]] to a [[Table]]. The field names will be taken from the field
- * names of the input type.
- *
- * Example:
- *
- * {{{
- * val in: DataSet[(String, Int)] = ...
- * val table = in.toTable
- * }}}
- *
- * This results in a [[Table]] that has field `_1` of type `String` and field `_2`
- * of type `Int`.
- */
-
- def toTable: Table = {
- val resultFields = inputType.getFieldNames.map(UnresolvedFieldReference)
- as(resultFields: _*)
- }
-
-}
-
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaBatchTranslator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaBatchTranslator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaBatchTranslator.scala
index cdcf53e..1c453fa 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaBatchTranslator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaBatchTranslator.scala
@@ -18,10 +18,8 @@
package org.apache.flink.api.scala.table
-
-import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.calcite.rel.RelNode
import org.apache.flink.api.java.table.JavaBatchTranslator
-import org.apache.flink.api.table.expressions.Expression
import org.apache.flink.api.scala.wrap
import org.apache.flink.api.table.plan._
import org.apache.flink.api.table.Table
@@ -30,7 +28,6 @@ import org.apache.flink.api.scala.DataSet
import scala.reflect.ClassTag
-
/**
* [[PlanTranslator]] for creating [[Table]]s from Scala [[DataSet]]s and
* translating them back to Scala [[DataSet]]s.
@@ -41,28 +38,13 @@ class ScalaBatchTranslator extends PlanTranslator {
type Representation[A] = DataSet[A]
- def createTable[A](
- repr: DataSet[A],
- fields: Array[Expression]): Table = {
-
- val result = javaTranslator.createTable(repr.javaSet, fields)
-
- new Table(result.operation)
+ override def createTable[A](repr: Representation[A], fieldNames: Array[String]): Table = {
+ javaTranslator.createTable(repr.javaSet, fieldNames)
}
- override def translate[O](op: PlanNode)(implicit tpe: TypeInformation[O]): DataSet[O] = {
+ override def translate[O](op: RelNode)(implicit tpe: TypeInformation[O]): DataSet[O] = {
// fake it till you make it ...
wrap(javaTranslator.translate(op))(ClassTag.AnyRef.asInstanceOf[ClassTag[O]])
}
- override def createTable[A](
- repr: Representation[A],
- inputType: CompositeType[A],
- expressions: Array[Expression],
- resultFields: Seq[(String, TypeInformation[_])]): Table = {
-
- val result = javaTranslator.createTable(repr.javaSet, inputType, expressions, resultFields)
-
- Table(result.operation)
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaStreamingTranslator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaStreamingTranslator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaStreamingTranslator.scala
deleted file mode 100644
index 88f1b83..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaStreamingTranslator.scala
+++ /dev/null
@@ -1,58 +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
-
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.java.table.JavaStreamingTranslator
-import org.apache.flink.api.table.Table
-import org.apache.flink.api.table.plan._
-import org.apache.flink.api.table.expressions.Expression
-import org.apache.flink.streaming.api.scala.{DataStream, javaToScalaStream}
-
-/**
- * [[PlanTranslator]] for creating [[Table]]s from Scala [[DataStream]]s and
- * translating them back to Scala [[DataStream]]s.
- *
- * This is very limited right now. Only select and filter are implemented. Also, the expression
- * operations must be extended to allow windowing operations.
- */
-class ScalaStreamingTranslator extends PlanTranslator {
-
- private val javaTranslator = new JavaStreamingTranslator
-
- override type Representation[A] = DataStream[A]
-
- override def translate[O](op: PlanNode)(implicit tpe: TypeInformation[O]): DataStream[O] = {
- // fake it till you make it ...
- javaToScalaStream(javaTranslator.translate(op))
- }
-
- override def createTable[A](
- repr: Representation[A],
- inputType: CompositeType[A],
- expressions: Array[Expression],
- resultFields: Seq[(String, TypeInformation[_])]): Table = {
-
- val result =
- javaTranslator.createTable(repr.getJavaStream, inputType, expressions, resultFields)
-
- new Table(result.operation)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableConversions.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableConversions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableConversions.scala
index 4f2172e..fdcd804 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableConversions.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableConversions.scala
@@ -22,10 +22,8 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.scala._
import org.apache.flink.api.table._
-import org.apache.flink.streaming.api.scala.DataStream
-
/**
- * Methods for converting a [[Table]] to a [[DataSet]] or [[DataStream]]. A [[Table]] is
+ * Methods for converting a [[Table]] to a [[DataSet]]. A [[Table]] is
* wrapped in this by the implicit conversions in [[org.apache.flink.api.scala.table]].
*/
class TableConversions(table: Table) {
@@ -34,14 +32,8 @@ class TableConversions(table: Table) {
* Converts the [[Table]] to a [[DataSet]].
*/
def toDataSet[T: TypeInformation]: DataSet[T] = {
- new ScalaBatchTranslator().translate[T](table.operation)
+ new ScalaBatchTranslator().translate[T](table.relNode)
}
- /**
- * Converts the [[Table]] to a [[DataStream]].
- */
- def toDataStream[T: TypeInformation]: DataStream[T] = {
- new ScalaStreamingTranslator().translate[T](table.operation)
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
index 0be6be2..058ff0e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
@@ -55,11 +55,6 @@ trait ImplicitExpressionOperations {
def * (other: Expression) = Mul(expr, other)
def % (other: Expression) = Mod(expr, other)
- def & (other: Expression) = BitwiseAnd(expr, other)
- def | (other: Expression) = BitwiseOr(expr, other)
- def ^ (other: Expression) = BitwiseXor(expr, other)
- def unary_~ = BitwiseNot(expr)
-
def abs = Abs(expr)
def sum = Sum(expr)
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/package.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/package.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/package.scala
index e74651b..86bb7c0 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/package.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/package.scala
@@ -19,7 +19,6 @@ package org.apache.flink.api.scala
import org.apache.flink.api.common.typeutils.CompositeType
import org.apache.flink.api.table.{Row, Table}
-import org.apache.flink.streaming.api.scala.DataStream
import scala.language.implicitConversions
@@ -32,7 +31,7 @@ import scala.language.implicitConversions
* import org.apache.flink.api.scala.table._
* }}}
*
- * imports implicit conversions for converting a [[DataSet]] or [[DataStream]] to a
+ * imports implicit conversions for converting a [[DataSet]] to a
* [[Table]]. This can be used to perform SQL-like queries on data. Please have
* a look at [[Table]] to see which operations are supported and
* [[org.apache.flink.api.scala.table.ImplicitExpressionOperations]] to see how an
@@ -78,7 +77,7 @@ package object table extends ImplicitExpressionConversions {
implicit def table2RowDataSet(
table: Table): DataSet[Row] = {
- new ScalaBatchTranslator().translate[Row](table.operation)
+ new ScalaBatchTranslator().translate[Row](table.relNode)
}
implicit def rowDataSet2Table(
@@ -86,20 +85,4 @@ package object table extends ImplicitExpressionConversions {
rowDataSet.toTable
}
- implicit def dataStream2DataSetConversions[T](
- stream: DataStream[T]): DataStreamConversions[T] = {
- new DataStreamConversions[T](
- stream,
- stream.getJavaStream.getType.asInstanceOf[CompositeType[T]])
- }
-
- implicit def table2RowDataStream(
- table: Table): DataStream[Row] = {
- new ScalaStreamingTranslator().translate[Row](table.operation)
- }
-
- implicit def rowDataStream2Table(
- rowDataStream: DataStream[Row]): Table = {
- rowDataStream.toTable
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Table.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Table.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Table.scala
index 641f2fa..271aa99 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Table.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Table.scala
@@ -17,14 +17,22 @@
*/
package org.apache.flink.api.table
-import org.apache.flink.api.java.io.DiscardingOutputFormat
-import org.apache.flink.api.table.explain.PlanJsonParser
-import org.apache.flink.api.table.expressions.analysis.{GroupByAnalyzer, PredicateAnalyzer, SelectionAnalyzer}
-import org.apache.flink.api.table.expressions.{Expression, ResolvedFieldReference, UnresolvedFieldReference}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.`type`.RelDataTypeField
+import org.apache.calcite.rel.core.JoinRelType
+import org.apache.calcite.rex.RexNode
+import org.apache.calcite.tools.RelBuilder
+import org.apache.calcite.tools.RelBuilder.{AggCall, GroupKey}
+import org.apache.flink.api.table.plan.RexNodeTranslator
+import RexNodeTranslator.{toRexNode, extractAggCalls}
+import org.apache.flink.api.table.expressions.Expression
import org.apache.flink.api.table.parser.ExpressionParser
-import org.apache.flink.api.table.plan._
-import org.apache.flink.api.scala._
-import org.apache.flink.api.scala.table._
+
+import scala.collection.JavaConverters._
+
+case class BaseTable(
+ private[flink] val relNode: RelNode,
+ private[flink] val relBuilder: RelBuilder)
/**
* The abstraction for writing Table API programs. Similar to how the batch and streaming APIs
@@ -50,7 +58,11 @@ import org.apache.flink.api.scala.table._
* in a Scala DSL or as an expression String. Please refer to the documentation for the expression
* syntax.
*/
-case class Table(private[flink] val operation: PlanNode) {
+class Table(
+ private[flink] override val relNode: RelNode,
+ private[flink] override val relBuilder: RelBuilder)
+ extends BaseTable(relNode, relBuilder)
+{
/**
* Performs a selection operation. Similar to an SQL SELECT statement. The field expressions
@@ -63,14 +75,30 @@ case class Table(private[flink] val operation: PlanNode) {
* }}}
*/
def select(fields: Expression*): Table = {
- val analyzer = new SelectionAnalyzer(operation.outputFields)
- val analyzedFields = fields.map(analyzer.analyze)
- val fieldNames = analyzedFields map(_.name)
- if (fieldNames.toSet.size != fieldNames.size) {
- throw new ExpressionException(s"Resulting fields names are not unique in expression" +
- s""" "${fields.mkString(", ")}".""")
+
+ relBuilder.push(relNode)
+
+ // separate aggregations and selection expressions
+ val extractedAggCalls: List[(Expression, List[AggCall])] = fields
+ .map(extractAggCalls(_, relBuilder)).toList
+
+ // get aggregation calls
+ val aggCalls: List[AggCall] = extractedAggCalls
+ .map(_._2).reduce( (x,y) => x ::: y)
+
+ // apply aggregations
+ if (aggCalls.nonEmpty) {
+ val emptyKey: GroupKey = relBuilder.groupKey()
+ relBuilder.aggregate(emptyKey, aggCalls.toIterable.asJava)
}
- this.copy(operation = Select(operation, analyzedFields))
+
+ // get selection expressions
+ val exprs: List[RexNode] = extractedAggCalls
+ .map(_._1)
+ .map(toRexNode(_, relBuilder))
+
+ relBuilder.project(exprs.toIterable.asJava)
+ new Table(relBuilder.build(), relBuilder)
}
/**
@@ -99,13 +127,12 @@ case class Table(private[flink] val operation: PlanNode) {
* }}}
*/
def as(fields: Expression*): Table = {
- fields forall {
- f => f.isInstanceOf[UnresolvedFieldReference]
- } match {
- case true =>
- case false => throw new ExpressionException("Only field expression allowed in as().")
- }
- this.copy(operation = As(operation, fields.toArray map { _.name }))
+
+ relBuilder.push(relNode)
+ val expressions = fields.map(toRexNode(_, relBuilder)).toIterable.asJava
+ val names = fields.map(_.name).toIterable.asJava
+ relBuilder.project(expressions, names)
+ new Table(relBuilder.build(), relBuilder)
}
/**
@@ -134,9 +161,11 @@ case class Table(private[flink] val operation: PlanNode) {
* }}}
*/
def filter(predicate: Expression): Table = {
- val analyzer = new PredicateAnalyzer(operation.outputFields)
- val analyzedPredicate = analyzer.analyze(predicate)
- this.copy(operation = Filter(operation, analyzedPredicate))
+
+ relBuilder.push(relNode)
+ val pred = toRexNode(predicate, relBuilder)
+ relBuilder.filter(pred)
+ new Table(relBuilder.build(), relBuilder)
}
/**
@@ -192,20 +221,13 @@ case class Table(private[flink] val operation: PlanNode) {
* in.groupBy('key).select('key, 'value.avg)
* }}}
*/
- def groupBy(fields: Expression*): Table = {
- val analyzer = new GroupByAnalyzer(operation.outputFields)
- val analyzedFields = fields.map(analyzer.analyze)
+ def groupBy(fields: Expression*): GroupedTable = {
- val illegalKeys = analyzedFields filter {
- case fe: ResolvedFieldReference => false // OK
- case e => true
- }
+ relBuilder.push(relNode)
+ val groupExpr = fields.map(toRexNode(_, relBuilder)).toIterable.asJava
+ val groupKey = relBuilder.groupKey(groupExpr)
- if (illegalKeys.nonEmpty) {
- throw new ExpressionException("Illegal key expressions: " + illegalKeys.mkString(", "))
- }
-
- this.copy(operation = GroupBy(operation, analyzedFields))
+ new GroupedTable(relBuilder.build(), relBuilder, groupKey)
}
/**
@@ -218,7 +240,7 @@ case class Table(private[flink] val operation: PlanNode) {
* in.groupBy("key").select("key, value.avg")
* }}}
*/
- def groupBy(fields: String): Table = {
+ def groupBy(fields: String): GroupedTable = {
val fieldsExpr = ExpressionParser.parseExpressionList(fields)
groupBy(fieldsExpr: _*)
}
@@ -235,16 +257,21 @@ case class Table(private[flink] val operation: PlanNode) {
* }}}
*/
def join(right: Table): Table = {
- val leftInputNames = operation.outputFields.map(_._1).toSet
- val rightInputNames = right.operation.outputFields.map(_._1).toSet
- if (leftInputNames.intersect(rightInputNames).nonEmpty) {
- throw new ExpressionException(
- "Overlapping fields names on join input, result would be ambiguous: " +
- operation.outputFields.mkString(", ") +
- " and " +
- right.operation.outputFields.mkString(", ") )
+
+ // check that join inputs do not have overlapping field names
+ val leftFields = relNode.getRowType.getFieldNames.asScala.toSet
+ val rightFields = right.relNode.getRowType.getFieldNames.asScala.toSet
+ if (leftFields.intersect(rightFields).nonEmpty) {
+ throw new IllegalArgumentException("Overlapping fields names on join input.")
}
- this.copy(operation = Join(operation, right.operation))
+
+ relBuilder.push(relNode)
+ relBuilder.push(right.relNode)
+
+ relBuilder.join(JoinRelType.INNER, relBuilder.literal(true))
+ val join = relBuilder.build()
+ val rowT = join.getRowType()
+ new Table(join, relBuilder)
}
/**
@@ -258,17 +285,27 @@ case class Table(private[flink] val operation: PlanNode) {
* }}}
*/
def unionAll(right: Table): Table = {
- val leftInputFields = operation.outputFields
- val rightInputFields = right.operation.outputFields
- if (!leftInputFields.equals(rightInputFields)) {
- throw new ExpressionException(
- "The fields names of join inputs should be fully overlapped, left inputs fields:" +
- operation.outputFields.mkString(", ") +
- " and right inputs fields" +
- right.operation.outputFields.mkString(", ")
- )
+
+ val leftRowType: List[RelDataTypeField] = relNode.getRowType.getFieldList.asScala.toList
+ val rightRowType: List[RelDataTypeField] = right.relNode.getRowType.getFieldList.asScala.toList
+
+ if (leftRowType.length != rightRowType.length) {
+ throw new IllegalArgumentException("Unioned tables have varying row schema.")
+ }
+ else {
+ val zipped: List[(RelDataTypeField, RelDataTypeField)] = leftRowType.zip(rightRowType)
+ zipped.foreach { case (x, y) =>
+ if (!x.getName.equals(y.getName) || x.getType != y.getType) {
+ throw new IllegalArgumentException("Unioned tables have varying row schema.")
+ }
+ }
}
- this.copy(operation = UnionAll(operation, right.operation))
+
+ relBuilder.push(relNode)
+ relBuilder.push(right.relNode)
+
+ relBuilder.union(true)
+ new Table(relBuilder.build(), relBuilder)
}
/**
@@ -277,18 +314,79 @@ case class Table(private[flink] val operation: PlanNode) {
* referenced by the statement will be scanned.
*/
def explain(extended: Boolean): String = {
- val ast = operation
- val dataSet = this.toDataSet[Row]
- val env = dataSet.getExecutionEnvironment
- dataSet.output(new DiscardingOutputFormat[Row])
- val jasonSqlPlan = env.getExecutionPlan()
- val sqlPlan = PlanJsonParser.getSqlExecutionPlan(jasonSqlPlan, extended)
- val result = "== Abstract Syntax Tree ==\n" + ast + "\n\n" + "== Physical Execution Plan ==" +
- "\n" + sqlPlan
- return result
+
+ // TODO: enable once toDataSet() is working again
+
+// val ast = operation
+// val dataSet = this.toDataSet[Row]
+// val env = dataSet.getExecutionEnvironment
+// dataSet.output(new DiscardingOutputFormat[Row])
+// val jasonSqlPlan = env.getExecutionPlan()
+// val sqlPlan = PlanJsonParser.getSqlExecutionPlan(jasonSqlPlan, extended)
+// val result = "== Abstract Syntax Tree ==\n" + ast + "\n\n" + "== Physical Execution Plan ==" +
+// "\n" + sqlPlan
+// return result
+
+ ""
}
-
+
def explain(): String = explain(false)
-
- override def toString: String = s"Expression($operation)"
+}
+
+class GroupedTable(
+ private[flink] override val relNode: RelNode,
+ private[flink] override val relBuilder: RelBuilder,
+ private[flink] val groupKey: GroupKey) extends BaseTable(relNode, relBuilder) {
+
+ /**
+ * Performs a selection operation. Similar to an SQL SELECT statement. The field expressions
+ * can contain complex expressions and aggregations.
+ *
+ * Example:
+ *
+ * {{{
+ * in.select('key, 'value.avg + " The average" as 'average, 'other.substring(0, 10))
+ * }}}
+ */
+ def select(fields: Expression*): Table = {
+
+ relBuilder.push(relNode)
+
+ // separate aggregations and selection expressions
+ val extractedAggCalls: List[(Expression, List[AggCall])] = fields
+ .map(extractAggCalls(_, relBuilder)).toList
+
+ // get aggregation calls
+ val aggCalls: List[AggCall] = extractedAggCalls
+ .map(_._2).reduce( (x,y) => x ::: y)
+
+ // apply aggregations
+ if (aggCalls.nonEmpty) {
+ relBuilder.aggregate(groupKey, aggCalls.toIterable.asJava)
+ }
+
+ // get selection expressions
+ val exprs: List[RexNode] = extractedAggCalls
+ .map(_._1)
+ .map(toRexNode(_, relBuilder))
+
+ relBuilder.project(exprs.toIterable.asJava)
+ new Table(relBuilder.build(), relBuilder)
+ }
+
+ /**
+ * Performs a selection operation. Similar to an SQL SELECT statement. The field expressions
+ * can contain complex expressions and aggregations.
+ *
+ * Example:
+ *
+ * {{{
+ * in.select("key, value.avg + " The average" as average, other.substring(0, 10)")
+ * }}}
+ */
+ def select(fields: String): Table = {
+ val fieldExprs = ExpressionParser.parseExpressionList(fields)
+ select(fieldExprs: _*)
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/ExpressionCodeGenerator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/ExpressionCodeGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/ExpressionCodeGenerator.scala
index a03ba61..9592f2e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/ExpressionCodeGenerator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/ExpressionCodeGenerator.scala
@@ -561,41 +561,6 @@ abstract class ExpressionCodeGenerator[R](
""".stripMargin
}
- case BitwiseAnd(left, right) =>
- generateIfNonNull(left, right, expr.typeInfo) {
- (leftTerm, rightTerm) => s"(int) $leftTerm & (int) $rightTerm"
- }
-
- case BitwiseOr(left, right) =>
- generateIfNonNull(left, right, expr.typeInfo) {
- (leftTerm, rightTerm) => s"(int) $leftTerm | (int) $rightTerm"
- }
-
- case BitwiseXor(left, right) =>
- generateIfNonNull(left, right, expr.typeInfo) {
- (leftTerm, rightTerm) => s"(int) $leftTerm ^ (int) $rightTerm"
- }
-
- case BitwiseNot(child) =>
- val childCode = generateExpression(child)
- if (nullCheck) {
- childCode.code +
- s"""
- |boolean $nullTerm = ${childCode.nullTerm};
- |$resultTpe $resultTerm;
- |if ($nullTerm) {
- | $resultTerm = ${defaultPrimitive(child.typeInfo)};
- |} else {
- | $resultTerm = ~((int) ${childCode.resultTerm});
- |}
- """.stripMargin
- } else {
- childCode.code +
- s"""
- |$resultTpe $resultTerm = ~((int) ${childCode.resultTerm});
- """.stripMargin
- }
-
case Not(child) =>
val childCode = generateExpression(child)
if (nullCheck) {
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/aggregations.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/aggregations.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/aggregations.scala
index 08e319d..d2fbdff 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/aggregations.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/aggregations.scala
@@ -17,9 +17,8 @@
*/
package org.apache.flink.api.table.expressions
-import org.apache.flink.api.table.ExpressionException
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
-import org.apache.flink.api.java.aggregation.Aggregations
+import org.apache.flink.api.table.ExpressionException
abstract sealed class Aggregation extends UnaryExpression { self: Product =>
@@ -39,61 +38,24 @@ abstract sealed class Aggregation extends UnaryExpression { self: Product =>
}
override def toString = s"Aggregate($child)"
-
- def getIntermediateFields: Seq[Expression]
- def getFinalField(inputs: Seq[Expression]): Expression
- def getAggregations: Seq[Aggregations]
}
case class Sum(child: Expression) extends Aggregation {
override def toString = s"($child).sum"
-
- override def getIntermediateFields: Seq[Expression] = Seq(child)
- override def getFinalField(inputs: Seq[Expression]): Expression = inputs(0)
- override def getAggregations = Seq(Aggregations.SUM)
}
case class Min(child: Expression) extends Aggregation {
override def toString = s"($child).min"
-
- override def getIntermediateFields: Seq[Expression] = Seq(child)
- override def getFinalField(inputs: Seq[Expression]): Expression = inputs(0)
- override def getAggregations = Seq(Aggregations.MIN)
-
}
case class Max(child: Expression) extends Aggregation {
override def toString = s"($child).max"
-
- override def getIntermediateFields: Seq[Expression] = Seq(child)
- override def getFinalField(inputs: Seq[Expression]): Expression = inputs(0)
- override def getAggregations = Seq(Aggregations.MAX)
}
case class Count(child: Expression) extends Aggregation {
- override def typeInfo = {
- child.typeInfo match {
- case _ => // we can count anything... :D
- }
- BasicTypeInfo.INT_TYPE_INFO
- }
-
override def toString = s"($child).count"
-
- override def getIntermediateFields: Seq[Expression] = Seq(Literal(Integer.valueOf(1)))
- override def getFinalField(inputs: Seq[Expression]): Expression = inputs(0)
- override def getAggregations = Seq(Aggregations.SUM)
-
}
case class Avg(child: Expression) extends Aggregation {
override def toString = s"($child).avg"
-
- override def getIntermediateFields: Seq[Expression] = Seq(child, Literal(1))
- // This is just sweet. Use our own AST representation and let the code generator do
- // our dirty work.
- override def getFinalField(inputs: Seq[Expression]): Expression =
- Div(inputs(0), inputs(1))
- override def getAggregations = Seq(Aggregations.SUM, Aggregations.SUM)
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/ExtractEquiJoinFields.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/ExtractEquiJoinFields.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/ExtractEquiJoinFields.scala
deleted file mode 100644
index 797de55..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/ExtractEquiJoinFields.scala
+++ /dev/null
@@ -1,70 +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.expressions.analysis
-
-import org.apache.flink.api.table.expressions._
-import org.apache.flink.api.common.typeutils.CompositeType
-
-import scala.collection.mutable
-
-/**
- * Equi-join field extractor for Join Predicates and CoGroup predicates. The result is a modified
- * expression without the equi-join predicates together with indices of the join fields
- * from both the left and right input.
- */
-object ExtractEquiJoinFields {
- def apply(leftType: CompositeType[_], rightType: CompositeType[_], predicate: Expression) = {
-
- val joinFieldsLeft = mutable.MutableList[Int]()
- val joinFieldsRight = mutable.MutableList[Int]()
-
- val equiJoinExprs = mutable.MutableList[EqualTo]()
- // First get all `===` expressions that are not below an `Or`
- predicate.transformPre {
- case or@Or(_, _) => NopExpression()
- case eq@EqualTo(le: ResolvedFieldReference, re: ResolvedFieldReference) =>
- if (leftType.hasField(le.name) && rightType.hasField(re.name)) {
- joinFieldsLeft += leftType.getFieldIndex(le.name)
- joinFieldsRight += rightType.getFieldIndex(re.name)
- } else if (leftType.hasField(re.name) && rightType.hasField(le.name)) {
- joinFieldsLeft += leftType.getFieldIndex(re.name)
- joinFieldsRight += rightType.getFieldIndex(le.name)
- } else {
- // not an equi-join predicate
- }
- equiJoinExprs += eq
- eq
- }
-
- // then remove the equi join expressions from the predicate
- val resultExpr = predicate.transformPost {
- // For OR, we can eliminate the OR since the equi join
- // predicate is evaluated before the expression is evaluated
- case or@Or(NopExpression(), _) => NopExpression()
- case or@Or(_, NopExpression()) => NopExpression()
- // For AND we replace it with the other expression, since the
- // equi join predicate will always be true
- case and@And(NopExpression(), other) => other
- case and@And(other, NopExpression()) => other
- case eq : EqualTo if equiJoinExprs.contains(eq) =>
- NopExpression()
- }
-
- (resultExpr, joinFieldsLeft.toArray, joinFieldsRight.toArray)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/GroupByAnalyzer.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/GroupByAnalyzer.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/GroupByAnalyzer.scala
deleted file mode 100644
index 6c7ecb2..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/GroupByAnalyzer.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.expressions.analysis
-
-import org.apache.flink.api.table._
-import org.apache.flink.api.table.expressions.{ResolvedFieldReference, Expression}
-import org.apache.flink.api.common.typeinfo.TypeInformation
-
-import scala.collection.mutable
-
-import org.apache.flink.api.table.trees.{Rule, Analyzer}
-
-
-/**
- * Analyzer for grouping expressions. Only field expressions are allowed as grouping expressions.
- */
-class GroupByAnalyzer(inputFields: Seq[(String, TypeInformation[_])])
- extends Analyzer[Expression] {
-
- def rules = Seq(new ResolveFieldReferences(inputFields), CheckGroupExpression)
-
- object CheckGroupExpression extends Rule[Expression] {
-
- def apply(expr: Expression) = {
- val errors = mutable.MutableList[String]()
-
- expr match {
- case f: ResolvedFieldReference => // this is OK
- case other =>
- throw new ExpressionException(
- s"""Invalid grouping expression "$expr". Only field references are allowed.""")
- }
- expr
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/InsertAutoCasts.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/InsertAutoCasts.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/InsertAutoCasts.scala
deleted file mode 100644
index 0fdcab6..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/InsertAutoCasts.scala
+++ /dev/null
@@ -1,92 +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.expressions.analysis
-
-import org.apache.flink.api.table.expressions._
-import org.apache.flink.api.common.typeinfo.{IntegerTypeInfo, BasicTypeInfo}
-import org.apache.flink.api.table.trees.Rule
-
-/**
- * [[Rule]] that adds casts in arithmetic operations.
- */
-class InsertAutoCasts extends Rule[Expression] {
-
- def apply(expr: Expression) = {
- val result = expr.transformPost {
-
- case plus@Plus(o1, o2) =>
- // Plus is special case since we can cast anything to String for String concat
- if (o1.typeInfo != o2.typeInfo && o1.typeInfo.isBasicType && o2.typeInfo.isBasicType) {
- if (o1.typeInfo.asInstanceOf[BasicTypeInfo[_]].shouldAutocastTo(
- o2.typeInfo.asInstanceOf[BasicTypeInfo[_]])) {
- Plus(Cast(o1, o2.typeInfo), o2)
- } else if (o2.typeInfo.asInstanceOf[BasicTypeInfo[_]].shouldAutocastTo(
- o1.typeInfo.asInstanceOf[BasicTypeInfo[_]])) {
- Plus(o1, Cast(o2, o1.typeInfo))
- } else if (o1.typeInfo == BasicTypeInfo.STRING_TYPE_INFO) {
- Plus(o1, Cast(o2, BasicTypeInfo.STRING_TYPE_INFO))
- } else if (o2.typeInfo == BasicTypeInfo.STRING_TYPE_INFO) {
- Plus(Cast(o1, BasicTypeInfo.STRING_TYPE_INFO), o2)
- } else {
- plus
- }
- } else {
- plus
- }
-
- case ba: BinaryExpression if ba.isInstanceOf[BinaryArithmetic] ||
- ba.isInstanceOf[BinaryComparison] =>
- val o1 = ba.left
- val o2 = ba.right
- if (o1.typeInfo != o2.typeInfo && o1.typeInfo.isBasicType && o2.typeInfo.isBasicType) {
- if (o1.typeInfo.asInstanceOf[BasicTypeInfo[_]].shouldAutocastTo(
- o2.typeInfo.asInstanceOf[BasicTypeInfo[_]])) {
- ba.makeCopy(Seq(Cast(o1, o2.typeInfo), o2))
- } else if (o2.typeInfo.asInstanceOf[BasicTypeInfo[_]].shouldAutocastTo(
- o1.typeInfo.asInstanceOf[BasicTypeInfo[_]])) {
- ba.makeCopy(Seq(o1, Cast(o2, o1.typeInfo)))
- } else {
- ba
- }
- } else {
- ba
- }
-
- case ba: BinaryExpression if ba.isInstanceOf[BitwiseBinaryArithmetic] =>
- val o1 = ba.left
- val o2 = ba.right
- if (o1.typeInfo != o2.typeInfo && o1.typeInfo.isInstanceOf[IntegerTypeInfo[_]] &&
- o2.typeInfo.isInstanceOf[IntegerTypeInfo[_]]) {
- if (o1.typeInfo.asInstanceOf[BasicTypeInfo[_]].shouldAutocastTo(
- o2.typeInfo.asInstanceOf[BasicTypeInfo[_]])) {
- ba.makeCopy(Seq(Cast(o1, o2.typeInfo), o2))
- } else if (o2.typeInfo.asInstanceOf[BasicTypeInfo[_]].shouldAutocastTo(
- o1.typeInfo.asInstanceOf[BasicTypeInfo[_]])) {
- ba.makeCopy(Seq(o1, Cast(o2, o1.typeInfo)))
- } else {
- ba
- }
- } else {
- ba
- }
- }
-
- result
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/PredicateAnalyzer.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/PredicateAnalyzer.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/PredicateAnalyzer.scala
deleted file mode 100644
index e9236f7..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/PredicateAnalyzer.scala
+++ /dev/null
@@ -1,35 +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.expressions.analysis
-
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.table.expressions.Expression
-import org.apache.flink.api.table.trees.Analyzer
-
-/**
- * Analyzer for predicates, i.e. filter operations and where clauses of joins.
- */
-class PredicateAnalyzer(inputFields: Seq[(String, TypeInformation[_])])
- extends Analyzer[Expression] {
- def rules = Seq(
- new ResolveFieldReferences(inputFields),
- new InsertAutoCasts,
- new TypeCheck,
- new VerifyNoAggregates,
- new VerifyBoolean)
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/ResolveFieldReferences.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/ResolveFieldReferences.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/ResolveFieldReferences.scala
deleted file mode 100644
index db7ea6c..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/ResolveFieldReferences.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.expressions.analysis
-
-import org.apache.flink.api.table.expressions.{ResolvedFieldReference,
-UnresolvedFieldReference, Expression}
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.table._
-
-import scala.collection.mutable
-
-import org.apache.flink.api.table.trees.Rule
-
-/**
- * Rule that resolved field references. This rule verifies that field references point to existing
- * fields of the input operation and creates [[ResolvedFieldReference]]s that hold the field
- * [[TypeInformation]] in addition to the field name.
- */
-class ResolveFieldReferences(inputFields: Seq[(String, TypeInformation[_])])
- extends Rule[Expression] {
-
- def apply(expr: Expression) = {
- val errors = mutable.MutableList[String]()
-
- val result = expr.transformPost {
- case fe@UnresolvedFieldReference(fieldName) =>
- inputFields.find { _._1 == fieldName } match {
- case Some((_, tpe)) => ResolvedFieldReference(fieldName, tpe)
-
- case None =>
- errors +=
- s"Field '$fieldName' is not valid for input fields ${inputFields.mkString(",")}"
- fe
- }
- }
-
- if (errors.length > 0) {
- throw new ExpressionException(
- s"""Invalid expression "$expr": ${errors.mkString(" ")}""")
- }
-
- result
-
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/SelectionAnalyzer.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/SelectionAnalyzer.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/SelectionAnalyzer.scala
deleted file mode 100644
index 625fdbf..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/SelectionAnalyzer.scala
+++ /dev/null
@@ -1,36 +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.expressions.analysis
-
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.table.expressions.Expression
-import org.apache.flink.api.table.trees.Analyzer
-
-/**
- * This analyzes selection expressions.
- */
-class SelectionAnalyzer(inputFields: Seq[(String, TypeInformation[_])])
- extends Analyzer[Expression] {
-
- def rules = Seq(
- new ResolveFieldReferences(inputFields),
- new VerifyNoNestedAggregates,
- new InsertAutoCasts,
- new TypeCheck)
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/TypeCheck.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/TypeCheck.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/TypeCheck.scala
deleted file mode 100644
index b724561..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/TypeCheck.scala
+++ /dev/null
@@ -1,57 +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.expressions.analysis
-
-import org.apache.flink.api.table.expressions.Expression
-import org.apache.flink.api.table.trees.Rule
-import org.apache.flink.api.table.{_}
-
-import scala.collection.mutable
-
-/**
- * Rule that makes sure we call [[Expression.typeInfo]] on each [[Expression]] at least once.
- * Expressions are expected to perform type verification in this method.
- */
-class TypeCheck extends Rule[Expression] {
-
- def apply(expr: Expression) = {
- val errors = mutable.MutableList[String]()
-
- val result = expr.transformPre {
- case expr: Expression=> {
- // simply get the typeInfo from the expression. this will perform type analysis
- try {
- expr.typeInfo
- } catch {
- case e: ExpressionException =>
- errors += e.getMessage
- }
- expr
- }
- }
-
- if (errors.length > 0) {
- throw new ExpressionException(
- s"""Invalid expression "$expr": ${errors.mkString(" ")}""")
- }
-
- result
-
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/VerifyBoolean.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/VerifyBoolean.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/VerifyBoolean.scala
deleted file mode 100644
index e75dd20..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/VerifyBoolean.scala
+++ /dev/null
@@ -1,41 +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.expressions.analysis
-
-import org.apache.flink.api.table.expressions.{NopExpression, Expression}
-import org.apache.flink.api.table.trees.Rule
-import org.apache.flink.api.table.{_}
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo
-
-import scala.collection.mutable
-
-/**
- * [[Rule]] that verifies that the result type of an [[Expression]] is Boolean. This is required
- * for filter/join predicates.
- */
-class VerifyBoolean extends Rule[Expression] {
-
- def apply(expr: Expression) = {
- if (!expr.isInstanceOf[NopExpression] && expr.typeInfo != BasicTypeInfo.BOOLEAN_TYPE_INFO) {
- throw new ExpressionException(s"Expression $expr of type ${expr.typeInfo} is not boolean.")
- }
-
- expr
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e4ee0c6d/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/VerifyNoAggregates.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/VerifyNoAggregates.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/VerifyNoAggregates.scala
deleted file mode 100644
index 09dbf88..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/VerifyNoAggregates.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.expressions.analysis
-
-import org.apache.flink.api.table.ExpressionException
-import org.apache.flink.api.table.expressions.{Aggregation, Expression}
-
-import scala.collection.mutable
-
-import org.apache.flink.api.table.trees.Rule
-
-/**
- * Rule that verifies that an expression does not contain aggregate operations. Right now, join
- * predicates and filter predicates cannot contain aggregates.
- */
-class VerifyNoAggregates extends Rule[Expression] {
-
- def apply(expr: Expression) = {
- val errors = mutable.MutableList[String]()
-
- val result = expr.transformPre {
- case agg: Aggregation=> {
- errors +=
- s"""Aggregations are not allowed in join/filter predicates."""
- agg
- }
- }
-
- if (errors.length > 0) {
- throw new ExpressionException(
- s"""Invalid expression "$expr": ${errors.mkString(" ")}""")
- }
-
- result
-
- }
-}