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/11 15:45:09 UTC
flink git commit: [FLINK-3604][tableAPI] Enable and fix ignored tests.
Repository: flink
Updated Branches:
refs/heads/tableOnCalcite d720b002a -> 3f8cea74a
[FLINK-3604][tableAPI] Enable and fix ignored tests.
This closes #1782.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3f8cea74
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3f8cea74
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3f8cea74
Branch: refs/heads/tableOnCalcite
Commit: 3f8cea74a79d31ee1d0cb74a767635f70a29e3c1
Parents: d720b00
Author: Fabian Hueske <fh...@apache.org>
Authored: Thu Mar 10 14:59:18 2016 +0100
Committer: vasia <va...@apache.org>
Committed: Fri Mar 11 15:44:20 2016 +0100
----------------------------------------------------------------------
.../api/java/table/JavaBatchTranslator.scala | 6 +-
.../api/scala/table/DataSetConversions.scala | 14 +-
.../apache/flink/api/scala/table/package.scala | 2 +-
.../flink/api/table/plan/TypeConverter.scala | 23 +-
.../plan/nodes/dataset/DataSetAggregate.scala | 2 +-
.../org/apache/flink/api/table/table.scala | 361 ++++++++++---------
.../flink/api/table/typeinfo/RowTypeInfo.scala | 15 +-
.../api/java/table/test/AggregationsITCase.java | 8 +-
.../flink/api/java/table/test/AsITCase.java | 84 ++---
.../table/test/GroupedAggregationsITCase.java | 1 -
.../api/java/table/test/PojoGroupingITCase.java | 3 +-
.../table/test/StringExpressionsITCase.java | 16 +-
.../scala/table/test/PageRankTableITCase.java | 4 +-
.../scala/table/test/AggregationsITCase.scala | 16 +-
.../flink/api/scala/table/test/AsITCase.scala | 49 ++-
.../api/scala/table/test/CastingITCase.scala | 3 +-
.../api/scala/table/test/FilterITCase.scala | 35 +-
.../table/test/StringExpressionsITCase.scala | 13 +-
18 files changed, 316 insertions(+), 339 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/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 14ee78e..dbbe7e8 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
@@ -88,11 +88,13 @@ class JavaBatchTranslator(config: TableConfig) extends PlanTranslator {
s"Cannot generate a valid execution plan for the given query: \n\n" +
s"${RelOptUtil.toString(lPlan)}\n" +
"Please consider filing a bug report.", e)
+ case a: AssertionError =>
+ throw a.getCause
}
- println("---------------")
+ println("-------------")
println("DataSet Plan:")
- println("---------------")
+ println("-------------")
println(RelOptUtil.toString(dataSetPlan))
dataSetPlan match {
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataSetConversions.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataSetConversions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataSetConversions.scala
index 2508a3d..6e630ea 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataSetConversions.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataSetConversions.scala
@@ -17,6 +17,7 @@
*/
package org.apache.flink.api.scala.table
+import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.table._
import org.apache.flink.api.table.expressions.{UnresolvedFieldReference, Expression}
import org.apache.flink.api.common.typeutils.CompositeType
@@ -27,7 +28,7 @@ import org.apache.flink.api.scala._
* Methods for converting a [[DataSet]] to a [[Table]]. A [[DataSet]] is
* wrapped in this by the implicit conversions in [[org.apache.flink.api.scala.table]].
*/
-class DataSetConversions[T](set: DataSet[T], inputType: CompositeType[T]) {
+class DataSetConversions[T](set: DataSet[T], inputType: TypeInformation[T]) {
/**
* Converts the [[DataSet]] to a [[Table]]. The field names can be specified like this:
@@ -59,8 +60,15 @@ class DataSetConversions[T](set: DataSet[T], inputType: CompositeType[T]) {
* of type `Int`.
*/
def toTable: Table = {
- val resultFields = inputType.getFieldNames.map(UnresolvedFieldReference)
- as(resultFields: _*)
+
+ inputType match {
+ case c: CompositeType[T] =>
+ val resultFields = c.getFieldNames.map(UnresolvedFieldReference)
+ as(resultFields: _*)
+ case _ =>
+ throw new IllegalArgumentException("" +
+ "Please specify a field name with 'as' to convert an atomic type dataset to a table ")
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/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 86bb7c0..5c3ba44 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
@@ -72,7 +72,7 @@ package object table extends ImplicitExpressionConversions {
}
implicit def dataSet2DataSetConversions[T](set: DataSet[T]): DataSetConversions[T] = {
- new DataSetConversions[T](set, set.getType.asInstanceOf[CompositeType[T]])
+ new DataSetConversions[T](set, set.getType)
}
implicit def table2RowDataSet(
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TypeConverter.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TypeConverter.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TypeConverter.scala
index 62c87a1..030d577 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TypeConverter.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TypeConverter.scala
@@ -37,7 +37,7 @@ import scala.collection.JavaConversions._
object TypeConverter {
- val DEFAULT_ROW_TYPE = new RowTypeInfo(Seq()).asInstanceOf[TypeInformation[Any]]
+ val DEFAULT_ROW_TYPE = new RowTypeInfo(Seq(), Seq()).asInstanceOf[TypeInformation[Any]]
def typeInfoToSqlType(typeInfo: TypeInformation[_]): SqlTypeName = typeInfo match {
case BOOLEAN_TYPE_INFO => BOOLEAN
@@ -132,17 +132,16 @@ object TypeConverter {
// POJO type expected
case pt: PojoTypeInfo[_] =>
- logicalFieldTypes.zipWithIndex foreach {
- case (fieldTypeInfo, i) =>
- val fieldName = logicalFieldNames(i)
- val index = pt.getFieldIndex(fieldName)
- if (index < 0) {
- throw new TableException(s"POJO does not define field name: $fieldName")
+ logicalFieldNames.zip(logicalFieldTypes) foreach {
+ case (fName, fType) =>
+ val pojoIdx = pt.getFieldIndex(fName)
+ if (pojoIdx < 0) {
+ throw new TableException(s"POJO does not define field name: $fName")
}
- val expectedTypeInfo = pt.getTypeAt(i)
- if (fieldTypeInfo != expectedTypeInfo) {
+ val expectedTypeInfo = pt.getTypeAt(pojoIdx)
+ if (fType != expectedTypeInfo) {
throw new TableException(s"Result field does not match expected type. " +
- s"Expected: $expectedTypeInfo; Actual: $fieldTypeInfo")
+ s"Expected: $expectedTypeInfo; Actual: $fType")
}
}
@@ -172,7 +171,7 @@ object TypeConverter {
// Row is expected, create the arity for it
case Some(typeInfo) if typeInfo.getTypeClass == classOf[Row] =>
- new RowTypeInfo(logicalFieldTypes)
+ new RowTypeInfo(logicalFieldTypes, logicalFieldNames)
// no physical type
// determine type based on logical fields and configuration parameters
@@ -180,7 +179,7 @@ object TypeConverter {
// no need for efficient types -> use Row
// we cannot use efficient types if row arity > tuple arity or nullable
if (!useEfficientTypes || logicalFieldTypes.length > Tuple.MAX_ARITY || nullable) {
- new RowTypeInfo(logicalFieldTypes)
+ new RowTypeInfo(logicalFieldTypes, logicalFieldNames)
}
// use efficient type tuple or atomic type
else {
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
index 9a9bf99..d3416ee 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
@@ -91,7 +91,7 @@ class DataSetAggregate(
.map(n => TypeConverter.sqlTypeToTypeInfo(n))
.toArray
- val rowTypeInfo = new RowTypeInfo(fieldTypes)
+ val rowTypeInfo = new RowTypeInfo(fieldTypes, rowType.getFieldNames.asScala)
val mappedInput = inputDS.map(aggregateResult._1)
val groupReduceFunction = aggregateResult._2
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/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 0763f34..43c097e 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
@@ -28,39 +28,39 @@ import org.apache.calcite.tools.RelBuilder.{AggCall, GroupKey}
import org.apache.calcite.util.NlsString
import org.apache.flink.api.table.plan.{PlanGenException, RexNodeTranslator}
import RexNodeTranslator.{toRexNode, extractAggCalls}
-import org.apache.flink.api.table.expressions.Expression
+import org.apache.flink.api.table.expressions.{Naming, UnresolvedFieldReference, Expression}
import org.apache.flink.api.table.parser.ExpressionParser
import scala.collection.JavaConverters._
case class BaseTable(
- private[flink] val relNode: RelNode,
- private[flink] val relBuilder: RelBuilder)
+ 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
- * have [[org.apache.flink.api.scala.DataSet]] and
- * [[org.apache.flink.streaming.api.scala.DataStream]].
- *
- * Use the methods of [[Table]] to transform data. Use
- * [[org.apache.flink.api.java.table.TableEnvironment]] to convert a [[Table]] back to a DataSet
- * or DataStream.
- *
- * When using Scala a [[Table]] can also be converted using implicit conversions.
- *
- * Example:
- *
- * {{{
- * val table = set.toTable('a, 'b)
- * ...
- * val table2 = ...
- * val set = table2.toDataSet[MyType]
- * }}}
- *
- * Operations such as [[join]], [[select]], [[where]] and [[groupBy]] either take arguments
- * in a Scala DSL or as an expression String. Please refer to the documentation for the expression
- * syntax.
- */
+ * The abstraction for writing Table API programs. Similar to how the batch and streaming APIs
+ * have [[org.apache.flink.api.scala.DataSet]] and
+ * [[org.apache.flink.streaming.api.scala.DataStream]].
+ *
+ * Use the methods of [[Table]] to transform data. Use
+ * [[org.apache.flink.api.java.table.TableEnvironment]] to convert a [[Table]] back to a DataSet
+ * or DataStream.
+ *
+ * When using Scala a [[Table]] can also be converted using implicit conversions.
+ *
+ * Example:
+ *
+ * {{{
+ * val table = set.toTable('a, 'b)
+ * ...
+ * val table2 = ...
+ * val set = table2.toDataSet[MyType]
+ * }}}
+ *
+ * Operations such as [[join]], [[select]], [[where]] and [[groupBy]] either take arguments
+ * in a Scala DSL or as an expression String. Please refer to the documentation for the expression
+ * syntax.
+ */
class Table(
private[flink] override val relNode: RelNode,
private[flink] override val relBuilder: RelBuilder)
@@ -68,15 +68,15 @@ class Table(
{
/**
- * 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))
- * }}}
- */
+ * 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)
@@ -101,90 +101,95 @@ class Table(
.map(toRexNode(_, relBuilder))
relBuilder.project(exprs.toIterable.asJava)
- var projected = relBuilder.build()
+ val projected = relBuilder.build()
if(relNode == projected) {
// Calcite's RelBuilder does not translate identity projects even if they rename fields.
// Add a projection ourselves (will be automatically removed by translation rules).
- val names = exprs.map{ e =>
- e.getKind match {
- case SqlKind.AS =>
- e.asInstanceOf[RexCall].getOperands.get(1)
- .asInstanceOf[RexLiteral].getValue
- .asInstanceOf[NlsString].getValue
- case SqlKind.INPUT_REF =>
- relNode.getRowType.getFieldNames.get(e.asInstanceOf[RexInputRef].getIndex)
- case _ =>
- throw new PlanGenException("Unexpected expression type encountered.")
- }
-
- }
-
- projected = LogicalProject.create(relNode, exprs.toList.asJava, names.toList.asJava)
+ new Table(createRenamingProject(exprs), relBuilder)
+ } else {
+ new Table(projected, relBuilder)
}
- new Table(projected, 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)")
- * }}}
- */
+ * 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: _*)
}
/**
- * Renames the fields of the expression result. Use this to disambiguate fields before
- * joining to operations.
- *
- * Example:
- *
- * {{{
- * in.as('a, 'b)
- * }}}
- */
+ * Renames the fields of the expression result. Use this to disambiguate fields before
+ * joining to operations.
+ *
+ * Example:
+ *
+ * {{{
+ * in.as('a, 'b)
+ * }}}
+ */
def as(fields: Expression*): Table = {
+ val curNames = relNode.getRowType.getFieldNames.asScala
+
+ // validate that AS has only field references
+ if (! fields.forall( _.isInstanceOf[UnresolvedFieldReference] )) {
+ throw new IllegalArgumentException("All expressions must be field references.")
+ }
+ // validate that we have not more field references than fields
+ if ( fields.length > curNames.size) {
+ throw new IllegalArgumentException("More field references than fields.")
+ }
+
+ val curFields = curNames.map(new UnresolvedFieldReference(_))
+
+ val renamings = fields.zip(curFields).map {
+ case (newName, oldName) => new Naming(oldName, newName.name)
+ }
+ val remaining = curFields.drop(fields.size)
+
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)
+
+ val exprs = (renamings ++ remaining).map(toRexNode(_, relBuilder))
+
+ new Table(createRenamingProject(exprs), relBuilder)
}
/**
- * Renames the fields of the expression result. Use this to disambiguate fields before
- * joining to operations.
- *
- * Example:
- *
- * {{{
- * in.as("a, b")
- * }}}
- */
+ * Renames the fields of the expression result. Use this to disambiguate fields before
+ * joining to operations.
+ *
+ * Example:
+ *
+ * {{{
+ * in.as("a, b")
+ * }}}
+ */
def as(fields: String): Table = {
val fieldExprs = ExpressionParser.parseExpressionList(fields)
as(fieldExprs: _*)
}
/**
- * Filters out elements that don't pass the filter predicate. Similar to a SQL WHERE
- * clause.
- *
- * Example:
- *
- * {{{
- * in.filter('name === "Fred")
- * }}}
- */
+ * Filters out elements that don't pass the filter predicate. Similar to a SQL WHERE
+ * clause.
+ *
+ * Example:
+ *
+ * {{{
+ * in.filter('name === "Fred")
+ * }}}
+ */
def filter(predicate: Expression): Table = {
relBuilder.push(relNode)
@@ -194,58 +199,58 @@ class Table(
}
/**
- * Filters out elements that don't pass the filter predicate. Similar to a SQL WHERE
- * clause.
- *
- * Example:
- *
- * {{{
- * in.filter("name = 'Fred'")
- * }}}
- */
+ * Filters out elements that don't pass the filter predicate. Similar to a SQL WHERE
+ * clause.
+ *
+ * Example:
+ *
+ * {{{
+ * in.filter("name = 'Fred'")
+ * }}}
+ */
def filter(predicate: String): Table = {
val predicateExpr = ExpressionParser.parseExpression(predicate)
filter(predicateExpr)
}
/**
- * Filters out elements that don't pass the filter predicate. Similar to a SQL WHERE
- * clause.
- *
- * Example:
- *
- * {{{
- * in.where('name === "Fred")
- * }}}
- */
+ * Filters out elements that don't pass the filter predicate. Similar to a SQL WHERE
+ * clause.
+ *
+ * Example:
+ *
+ * {{{
+ * in.where('name === "Fred")
+ * }}}
+ */
def where(predicate: Expression): Table = {
filter(predicate)
}
/**
- * Filters out elements that don't pass the filter predicate. Similar to a SQL WHERE
- * clause.
- *
- * Example:
- *
- * {{{
- * in.where("name = 'Fred'")
- * }}}
- */
+ * Filters out elements that don't pass the filter predicate. Similar to a SQL WHERE
+ * clause.
+ *
+ * Example:
+ *
+ * {{{
+ * in.where("name = 'Fred'")
+ * }}}
+ */
def where(predicate: String): Table = {
filter(predicate)
}
/**
- * Groups the elements on some grouping keys. Use this before a selection with aggregations
- * to perform the aggregation on a per-group basis. Similar to a SQL GROUP BY statement.
- *
- * Example:
- *
- * {{{
- * in.groupBy('key).select('key, 'value.avg)
- * }}}
- */
+ * Groups the elements on some grouping keys. Use this before a selection with aggregations
+ * to perform the aggregation on a per-group basis. Similar to a SQL GROUP BY statement.
+ *
+ * Example:
+ *
+ * {{{
+ * in.groupBy('key).select('key, 'value.avg)
+ * }}}
+ */
def groupBy(fields: Expression*): GroupedTable = {
relBuilder.push(relNode)
@@ -256,29 +261,29 @@ class Table(
}
/**
- * Groups the elements on some grouping keys. Use this before a selection with aggregations
- * to perform the aggregation on a per-group basis. Similar to a SQL GROUP BY statement.
- *
- * Example:
- *
- * {{{
- * in.groupBy("key").select("key, value.avg")
- * }}}
- */
+ * Groups the elements on some grouping keys. Use this before a selection with aggregations
+ * to perform the aggregation on a per-group basis. Similar to a SQL GROUP BY statement.
+ *
+ * Example:
+ *
+ * {{{
+ * in.groupBy("key").select("key, value.avg")
+ * }}}
+ */
def groupBy(fields: String): GroupedTable = {
val fieldsExpr = ExpressionParser.parseExpressionList(fields)
groupBy(fieldsExpr: _*)
}
/**
- * Removes duplicate values and returns only distinct (different) values.
- *
- * Example:
- *
- * {{{
- * in.select("key, value").distinct()
- * }}}
- */
+ * Removes duplicate values and returns only distinct (different) values.
+ *
+ * Example:
+ *
+ * {{{
+ * in.select("key, value").distinct()
+ * }}}
+ */
def distinct(): Table = {
relBuilder.push(relNode)
relBuilder.distinct()
@@ -286,16 +291,16 @@ class Table(
}
/**
- * Joins two [[Table]]s. Similar to an SQL join. The fields of the two joined
- * operations must not overlap, use [[as]] to rename fields if necessary. You can use
- * where and select clauses after a join to further specify the behaviour of the join.
- *
- * Example:
- *
- * {{{
- * left.join(right).where('a === 'b && 'c > 3).select('a, 'b, 'd)
- * }}}
- */
+ * Joins two [[Table]]s. Similar to an SQL join. The fields of the two joined
+ * operations must not overlap, use [[as]] to rename fields if necessary. You can use
+ * where and select clauses after a join to further specify the behaviour of the join.
+ *
+ * Example:
+ *
+ * {{{
+ * left.join(right).where('a === 'b && 'c > 3).select('a, 'b, 'd)
+ * }}}
+ */
def join(right: Table): Table = {
// check that join inputs do not have overlapping field names
@@ -314,15 +319,15 @@ class Table(
}
/**
- * Union two [[Table]]s. Similar to an SQL UNION ALL. The fields of the two union operations
- * must fully overlap.
- *
- * Example:
- *
- * {{{
- * left.unionAll(right)
- * }}}
- */
+ * Union two [[Table]]s. Similar to an SQL UNION ALL. The fields of the two union operations
+ * must fully overlap.
+ *
+ * Example:
+ *
+ * {{{
+ * left.unionAll(right)
+ * }}}
+ */
def unionAll(right: Table): Table = {
val leftRowType: List[RelDataTypeField] = relNode.getRowType.getFieldList.asScala.toList
@@ -370,12 +375,32 @@ class Table(
}
def explain(): String = explain(false)
+
+ private def createRenamingProject(exprs: Seq[RexNode]): LogicalProject = {
+
+ val names = exprs.map{ e =>
+ e.getKind match {
+ case SqlKind.AS =>
+ e.asInstanceOf[RexCall].getOperands.get(1)
+ .asInstanceOf[RexLiteral].getValue
+ .asInstanceOf[NlsString].getValue
+ case SqlKind.INPUT_REF =>
+ relNode.getRowType.getFieldNames.get(e.asInstanceOf[RexInputRef].getIndex)
+ case _ =>
+ throw new PlanGenException("Unexpected expression type encountered.")
+ }
+
+ }
+
+ LogicalProject.create(relNode, exprs.toList.asJava, names.toList.asJava)
+ }
+
}
class GroupedTable(
- private[flink] override val relNode: RelNode,
- private[flink] override val relBuilder: RelBuilder,
- private[flink] val groupKey: GroupKey) extends BaseTable(relNode, relBuilder) {
+ 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
@@ -404,7 +429,7 @@ class GroupedTable(
// get selection expressions
val exprs: List[RexNode] = try {
- extractedAggCalls
+ extractedAggCalls
.map(_._1)
.map(toRexNode(_, relBuilder))
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeinfo/RowTypeInfo.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeinfo/RowTypeInfo.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeinfo/RowTypeInfo.scala
index 522c7f3..81c3836 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeinfo/RowTypeInfo.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeinfo/RowTypeInfo.scala
@@ -30,14 +30,25 @@ import org.apache.flink.api.table.Row
/**
* TypeInformation for [[Row]].
*/
-class RowTypeInfo(fieldTypes: Seq[TypeInformation[_]])
+class RowTypeInfo(fieldTypes: Seq[TypeInformation[_]], fieldNames: Seq[String])
extends CaseClassTypeInfo[Row](
classOf[Row],
Array(),
fieldTypes,
- for (i <- fieldTypes.indices) yield "f" + i)
+ fieldNames)
{
+ if (fieldTypes.length != fieldNames.length) {
+ throw new IllegalArgumentException("Number of field types and names is different.")
+ }
+ if (fieldNames.length != fieldNames.toSet.size) {
+ throw new IllegalArgumentException("Field names are not unique.")
+ }
+
+ def this(fieldTypes: Seq[TypeInformation[_]]) = {
+ this(fieldTypes, for (i <- fieldTypes.indices) yield "f" + i)
+ }
+
/**
* Temporary variable for directly passing orders to comparators.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
index 30598c4..e26bc32 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
@@ -44,15 +44,13 @@ import org.apache.flink.api.java.table.TableEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple7;
+import org.apache.flink.api.table.plan.PlanGenException;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
-import scala.NotImplementedError;
-
import java.util.List;
@RunWith(Parameterized.class)
@@ -160,9 +158,7 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
compareResultAsText(results, expected);
}
- // Calcite does not eagerly check type compatibility
- @Ignore
- @Test(expected = ExpressionException.class)
+ @Test(expected = PlanGenException.class)
public void testNonWorkingDataTypes() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
index 628cbef..097339e 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
@@ -49,8 +49,9 @@ public class AsITCase extends TableProgramsTestBase {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = getJavaTableEnvironment();
- Table table =
- tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c");
+ Table table = tableEnv
+ .fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c")
+ .select("a, b, c");
DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
List<Row> results = ds.collect();
@@ -69,8 +70,9 @@ public class AsITCase extends TableProgramsTestBase {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = getJavaTableEnvironment();
- Table table =
- tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c");
+ Table table = tableEnv
+ .fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c")
+ .select("a, b, c");
TypeInformation<?> ti = new TupleTypeInfo<Tuple3<Integer, Long, String>>(
BasicTypeInfo.INT_TYPE_INFO,
@@ -99,8 +101,9 @@ public class AsITCase extends TableProgramsTestBase {
data.add(new Tuple4<>("lol", 2, 1.0, "Hi"));
data.add(new Tuple4<>("Test me", 4, 3.33, "Hello world"));
- Table table =
- tableEnv.fromDataSet(env.fromCollection(data), "a, b, c, d");
+ Table table = tableEnv
+ .fromDataSet(env.fromCollection(data), "a, b, c, d")
+ .select("a, b, c, d");
DataSet<SmallPojo2> ds = tableEnv.toDataSet(table, SmallPojo2.class);
List<SmallPojo2> results = ds.collect();
@@ -118,12 +121,13 @@ public class AsITCase extends TableProgramsTestBase {
data.add(new SmallPojo("Anna", 56, 10000.00, "Engineering"));
data.add(new SmallPojo("Lucy", 42, 6000.00, "HR"));
- Table table =
- tableEnv.fromDataSet(env.fromCollection(data),
+ Table table = tableEnv
+ .fromDataSet(env.fromCollection(data),
"department AS a, " +
"age AS b, " +
"salary AS c, " +
- "name AS d");
+ "name AS d")
+ .select("a, b, c, d");
DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
List<Row> results = ds.collect();
@@ -144,12 +148,13 @@ public class AsITCase extends TableProgramsTestBase {
data.add(new PrivateSmallPojo("Anna", 56, 10000.00, "Engineering"));
data.add(new PrivateSmallPojo("Lucy", 42, 6000.00, "HR"));
- Table table =
- tableEnv.fromDataSet(env.fromCollection(data),
+ Table table = tableEnv
+ .fromDataSet(env.fromCollection(data),
"department AS a, " +
"age AS b, " +
"salary AS c, " +
- "name AS d");
+ "name AS d")
+ .select("a, b, c, d");
DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
List<Row> results = ds.collect();
@@ -170,12 +175,13 @@ public class AsITCase extends TableProgramsTestBase {
data.add(new SmallPojo("Anna", 56, 10000.00, "Engineering"));
data.add(new SmallPojo("Lucy", 42, 6000.00, "HR"));
- Table table =
- tableEnv.fromDataSet(env.fromCollection(data),
+ Table table = tableEnv
+ .fromDataSet(env.fromCollection(data),
"department AS a, " +
"age AS b, " +
"salary AS c, " +
- "name AS d");
+ "name AS d")
+ .select("a, b, c, d");
DataSet<SmallPojo2> ds = tableEnv.toDataSet(table, SmallPojo2.class);
List<SmallPojo2> results = ds.collect();
@@ -196,12 +202,13 @@ public class AsITCase extends TableProgramsTestBase {
data.add(new PrivateSmallPojo("Anna", 56, 10000.00, "Engineering"));
data.add(new PrivateSmallPojo("Lucy", 42, 6000.00, "HR"));
- Table table =
- tableEnv.fromDataSet(env.fromCollection(data),
+ Table table = tableEnv
+ .fromDataSet(env.fromCollection(data),
"department AS a, " +
"age AS b, " +
"salary AS c, " +
- "name AS d");
+ "name AS d")
+ .select("a, b, c, d");
DataSet<PrivateSmallPojo2> ds = tableEnv.toDataSet(table, PrivateSmallPojo2.class);
List<PrivateSmallPojo2> results = ds.collect();
@@ -217,13 +224,7 @@ public class AsITCase extends TableProgramsTestBase {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = getJavaTableEnvironment();
- Table table =
- tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b");
-
- DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
- List<Row> results = ds.collect();
- String expected = "";
- compareResultAsText(results, expected);
+ tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b");
}
@Test(expected = IllegalArgumentException.class)
@@ -231,13 +232,7 @@ public class AsITCase extends TableProgramsTestBase {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = getJavaTableEnvironment();
- Table table =
- tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c, d");
-
- DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
- List<Row> results = ds.collect();
- String expected = "";
- compareResultAsText(results, expected);
+ tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c, d");
}
@Test(expected = IllegalArgumentException.class)
@@ -245,13 +240,7 @@ public class AsITCase extends TableProgramsTestBase {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = getJavaTableEnvironment();
- Table table =
- tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, b");
-
- DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
- List<Row> results = ds.collect();
- String expected = "";
- compareResultAsText(results, expected);
+ tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, b");
}
@Test(expected = IllegalArgumentException.class)
@@ -259,13 +248,7 @@ public class AsITCase extends TableProgramsTestBase {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = getJavaTableEnvironment();
- Table table =
- tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a + 1, b, c");
-
- DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
- List<Row> results = ds.collect();
- String expected = "";
- compareResultAsText(results, expected);
+ tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a + 1, b, c");
}
@Test(expected = IllegalArgumentException.class)
@@ -273,14 +256,7 @@ public class AsITCase extends TableProgramsTestBase {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = getJavaTableEnvironment();
- Table table =
- tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a as foo, b," +
- " c");
-
- DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
- List<Row> results = ds.collect();
- String expected = "";
- compareResultAsText(results, expected);
+ tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a as foo, b, c");
}
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
index b8a16b3..91d1976 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
@@ -29,7 +29,6 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
-import scala.NotImplementedError;
import java.util.List;
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java
index 993638d..b8b84f6 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java
@@ -26,7 +26,6 @@ import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.table.TableEnvironment;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.table.Table;
-import org.apache.flink.api.table.TableException;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -39,7 +38,7 @@ public class PojoGroupingITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test(expected = TableException.class)
+ @Test
public void testPojoGrouping() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
index 6b8f984..707ee66 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
@@ -80,9 +80,7 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
compareResultAsText(results, expected);
}
- // Calcite does eagerly check expression types
- @Ignore
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = CodeGenException.class)
public void testNonWorkingSubstring1() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -97,14 +95,10 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
.select("a.substring(0, b)");
DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
- List<Row> results = resultSet.collect();
- String expected = "";
- compareResultAsText(results, expected);
+ resultSet.collect();
}
- // Calcite does eagerly check expression types
- @Ignore
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = CodeGenException.class)
public void testNonWorkingSubstring2() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -119,8 +113,6 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
.select("a.substring(b, 15)");
DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
- List<Row> results = resultSet.collect();
- String expected = "";
- compareResultAsText(results, expected);
+ resultSet.collect();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/test/java/org/apache/flink/api/scala/table/test/PageRankTableITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/scala/table/test/PageRankTableITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/scala/table/test/PageRankTableITCase.java
index 55f1bde..a893d4d 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/scala/table/test/PageRankTableITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/scala/table/test/PageRankTableITCase.java
@@ -74,7 +74,9 @@ public class PageRankTableITCase extends JavaProgramTestBase {
tConfigs.add(config);
}
- // TODO: Disabling test until Table API is operational again
+ // TODO: Enable test again once:
+ // 1) complex types (long[]) can be shipped through Table API
+ // 2) abs function is available
// return toParameterList(tConfigs);
return new LinkedList<>();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AggregationsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AggregationsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AggregationsITCase.scala
index 00261c0..ad9a66d 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AggregationsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AggregationsITCase.scala
@@ -18,6 +18,7 @@
package org.apache.flink.api.scala.table.test
+import org.apache.flink.api.table.plan.PlanGenException
import org.apache.flink.api.table.{ExpressionException, Row}
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.table._
@@ -50,10 +51,6 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
val env = ExecutionEnvironment.getExecutionEnvironment
val t = CollectionDataSets.get3TupleDataSet(env).toTable
.select('foo.avg)
-
- val expected = ""
- val results = t.toDataSet[Row].collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test
@@ -124,17 +121,14 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
TestBaseUtils.compareResultAsText(result.asJava, expected)
}
- @Ignore // Calcite does not eagerly check types
- @Test(expected = classOf[ExpressionException])
+ @Test(expected = classOf[PlanGenException])
def testNonWorkingAggregationDataTypes(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = env.fromElements(("Hello", 1)).toTable
.select('_1.sum)
- val expected = ""
- val results = t.toDataSet[Row].collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
+ t.collect()
}
@Test(expected = classOf[IllegalArgumentException])
@@ -143,10 +137,6 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
val env = ExecutionEnvironment.getExecutionEnvironment
val t = env.fromElements(("Hello", 1)).toTable
.select('_2.sum.sum)
-
- val expected = ""
- val results = t.toDataSet[Row].collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AsITCase.scala
index f2120ef..9f9a3b4 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AsITCase.scala
@@ -43,7 +43,9 @@ class AsITCase(
def testAs(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
- val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
+ val t = CollectionDataSets.get3TupleDataSet(env)
+ .as('a, 'b, 'c)
+ .select('a, 'b, 'c)
val expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" +
"4,3,Hello world, how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" +
@@ -64,7 +66,9 @@ class AsITCase(
SomeCaseClass("Anna", 56, 10000.00, "Engineering"),
SomeCaseClass("Lucy", 42, 6000.00, "HR"))
- val t = env.fromCollection(data).as('a, 'b, 'c, 'd)
+ val t = env.fromCollection(data)
+ .as('a, 'b, 'c, 'd)
+ .select('a, 'b, 'c, 'd)
val expected: String =
"Peter,28,4000.0,Sales\n" +
@@ -83,7 +87,9 @@ class AsITCase(
SomeCaseClass("Anna", 56, 10000.00, "Engineering"),
SomeCaseClass("Lucy", 42, 6000.00, "HR"))
- val t = env.fromCollection(data).as('a, 'b, 'c, 'd)
+ val t = env.fromCollection(data)
+ .as('a, 'b, 'c, 'd)
+ .select('a, 'b, 'c, 'd)
val expected: String =
"SomeCaseClass(Peter,28,4000.0,Sales)\n" +
@@ -97,33 +103,24 @@ class AsITCase(
def testAsWithToFewFields(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
- val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b)
-
- val expected = "no"
- val results = t.toDataSet[Row](getConfig).collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val t = CollectionDataSets.get3TupleDataSet(env)
+ .as('a, 'b)
}
@Test(expected = classOf[IllegalArgumentException])
def testAsWithToManyFields(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
- val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c, 'd)
-
- val expected = "no"
- val results = t.toDataSet[Row](getConfig).collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val t = CollectionDataSets.get3TupleDataSet(env)
+ .as('a, 'b, 'c, 'd)
}
@Test(expected = classOf[IllegalArgumentException])
def testAsWithAmbiguousFields(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
- val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'b)
-
- val expected = "no"
- val results = t.toDataSet[Row](getConfig).collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val t = CollectionDataSets.get3TupleDataSet(env)
+ .as('a, 'b, 'b)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -131,11 +128,9 @@ class AsITCase(
val env = ExecutionEnvironment.getExecutionEnvironment
// as can only have field references
- val t = CollectionDataSets.get3TupleDataSet(env).as('a + 1, 'b, 'b)
-
- val expected = "no"
- val results = t.toDataSet[Row](getConfig).collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val t = CollectionDataSets.get3TupleDataSet(env)
+ .as('a, 'b, 'c)
+ .as('a + 1, 'b, 'c)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -143,11 +138,9 @@ class AsITCase(
val env = ExecutionEnvironment.getExecutionEnvironment
// as can only have field references
- val t = CollectionDataSets.get3TupleDataSet(env).as('a as 'foo, 'b, 'b)
-
- val expected = "no"
- val results = t.toDataSet[Row](getConfig).collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val t = CollectionDataSets.get3TupleDataSet(env)
+ .as('a, 'b, 'c)
+ .as('a as 'foo, 'b, 'c)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
index 8e11f76..111525f 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
@@ -66,9 +66,8 @@ class CastingITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mo
TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- // TODO support advanced String operations
- @Ignore
+ @Ignore // TODO support advanced String operations
@Test
def testAutoCastToString(): Unit = {
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala
index 0febd4d..ae8ebef 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala
@@ -74,20 +74,18 @@ class FilterITCase(
TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- // TODO test broken does not test Table API
- @Ignore
@Test
def testFilterOnStringTupleField(): Unit = {
/*
* Test filter on String tuple field.
*/
val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env)
- val filterDs = ds.filter( _._3.contains("world") )
+ val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
+ val filterDs = ds.filter( 'c.like("%world%") )
-// val expected = "(3,2,Hello world)\n" + "(4,3,Hello world, how are you?)\n"
-// val results = filterDs.toDataSet[Row](getConfig).collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "3,2,Hello world\n" + "4,3,Hello world, how are you?\n"
+ val results = filterDs.toDataSet[Row](getConfig).collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test
@@ -152,27 +150,21 @@ class FilterITCase(
TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- // These two not yet done, but are planned
- // TODO test broken does not test Table API
- @Ignore
@Test
def testFilterBasicType(): Unit = {
/*
* Test filter on basic type
*/
-
val env = ExecutionEnvironment.getExecutionEnvironment
val ds = CollectionDataSets.getStringDataSet(env)
- val filterDs = ds.filter( _.startsWith("H") )
+ val filterDs = ds.as('a).filter( 'a.like("H%") )
-// val expected = "Hi\n" + "Hello\n" + "Hello world\n" + "Hello world, how are you?\n"
-// val results = filterDs.toDataSet[Row](getConfig).collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "Hi\n" + "Hello\n" + "Hello world\n" + "Hello world, how are you?\n"
+ val results = filterDs.toDataSet[Row](getConfig).collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- // TODO test broken does not test Table API
- @Ignore
@Test
def testFilterOnCustomType(): Unit = {
/*
@@ -180,11 +172,12 @@ class FilterITCase(
*/
val env = ExecutionEnvironment.getExecutionEnvironment
val ds = CollectionDataSets.getCustomTypeDataSet(env)
- val filterDs = ds.filter( _.myString.contains("a") )
+ val filterDs = ds.as('myInt as 'i, 'myLong as 'l, 'myString as 's)
+ .filter( 's.like("%a%") )
-// val expected = "3,3,Hello world, how are you?\n" + "3,4,I am fine.\n" + "3,5,Luke Skywalker\n"
-// val results = filterDs.toDataSet[Row](getConfig).collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "3,3,Hello world, how are you?\n" + "3,4,I am fine.\n" + "3,5,Luke Skywalker\n"
+ val results = filterDs.toDataSet[Row](getConfig).collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3f8cea74/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala
index c4fc346..7977547 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala
@@ -21,6 +21,7 @@ package org.apache.flink.api.scala.table.test
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.table._
import org.apache.flink.api.table.Row
+import org.apache.flink.api.table.codegen.CodeGenException
import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
import org.junit._
@@ -54,32 +55,24 @@ class StringExpressionsITCase(mode: TestExecutionMode) extends MultipleProgramsT
TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- // Calcite does eagerly check expression types
- @Ignore
- @Test(expected = classOf[IllegalArgumentException])
+ @Test(expected = classOf[CodeGenException])
def testNonWorkingSubstring1(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = env.fromElements(("AAAA", 2.0), ("BBBB", 1.0)).as('a, 'b)
.select('a.substring(0, 'b))
- val expected = "AAA\nBB"
val results = t.toDataSet[Row].collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- // Calcite does eagerly check expression types
- @Ignore
- @Test(expected = classOf[IllegalArgumentException])
+ @Test(expected = classOf[CodeGenException])
def testNonWorkingSubstring2(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = env.fromElements(("AAAA", "c"), ("BBBB", "d")).as('a, 'b)
.select('a.substring('b, 15))
- val expected = "AAA\nBB"
val results = t.toDataSet[Row].collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
}