flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject flink git commit: [FLINK-3226] Add DataSet scan and conversion to DataSet[Row]
Date Thu, 04 Feb 2016 11:15:30 GMT
Repository: flink
Updated Branches:
  refs/heads/tableOnCalcite 297564646 -> 1f3eda984


[FLINK-3226] Add DataSet scan and  conversion to DataSet[Row]

This closes #1579.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/1f3eda98
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/1f3eda98
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/1f3eda98

Branch: refs/heads/tableOnCalcite
Commit: 1f3eda984473be87207c94ef8118c635a74b090a
Parents: 2975646
Author: Fabian Hueske <fhueske@apache.org>
Authored: Tue Feb 2 17:15:28 2016 +0100
Committer: Fabian Hueske <fhueske@apache.org>
Committed: Thu Feb 4 12:14:34 2016 +0100

----------------------------------------------------------------------
 .../api/java/table/JavaBatchTranslator.scala    |   2 +
 .../api/scala/table/ScalaBatchTranslator.scala  |   8 +-
 .../flink/api/table/plan/PlanTranslator.scala   |  87 ++++++++---
 .../flink/api/table/plan/TypeConverter.scala    |  15 ++
 .../plan/nodes/dataset/DataSetSource.scala      | 150 ++++++++++++++++++-
 .../plan/rules/dataset/DataSetScanRule.scala    |   8 +-
 .../api/table/plan/schema/DataSetTable.scala    |  41 ++---
 .../flink/api/java/table/test/AsITCase.java     |  63 ++++++--
 .../flink/api/java/table/test/FilterITCase.java |  15 +-
 .../flink/api/java/table/test/SelectITCase.java |  15 +-
 .../flink/api/scala/table/test/AsITCase.scala   |  40 ++++-
 .../api/scala/table/test/SelectITCase.scala     |  28 ++--
 12 files changed, 380 insertions(+), 92 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/1f3eda98/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 7e91190..f70f477 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
@@ -40,11 +40,13 @@ class JavaBatchTranslator extends PlanTranslator {
 
   override def createTable[A](
       repr: Representation[A],
+      fieldIndexes: Array[Int],
       fieldNames: Array[String]): Table = {
 
     // create table representation from DataSet
     val dataSetTable = new DataSetTable[A](
       repr.asInstanceOf[JavaDataSet[A]],
+      fieldIndexes,
       fieldNames
     )
 

http://git-wip-us.apache.org/repos/asf/flink/blob/1f3eda98/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 1c453fa..cc92c37 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
@@ -38,8 +38,12 @@ class ScalaBatchTranslator extends PlanTranslator {
 
   type Representation[A] = DataSet[A]
 
-  override def createTable[A](repr: Representation[A], fieldNames: Array[String]): Table
= {
-    javaTranslator.createTable(repr.javaSet, fieldNames)
+  override def createTable[A](
+    repr: Representation[A],
+    fieldIndexes: Array[Int],
+    fieldNames: Array[String]): Table =
+  {
+    javaTranslator.createTable(repr.javaSet, fieldIndexes, fieldNames)
   }
 
   override def translate[O](op: RelNode)(implicit tpe: TypeInformation[O]): DataSet[O] =
{

http://git-wip-us.apache.org/repos/asf/flink/blob/1f3eda98/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/PlanTranslator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/PlanTranslator.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/PlanTranslator.scala
index 4e97f83..af22768 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/PlanTranslator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/PlanTranslator.scala
@@ -18,10 +18,12 @@
 package org.apache.flink.api.table.plan
 
 import org.apache.calcite.rel.RelNode
-import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.common.typeinfo.{AtomicType, TypeInformation}
 import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfo}
+import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
 import org.apache.flink.api.table.parser.ExpressionParser
-import org.apache.flink.api.table.expressions.{Expression, ResolvedFieldReference, UnresolvedFieldReference}
+import org.apache.flink.api.table.expressions.{Naming, Expression, UnresolvedFieldReference}
 import org.apache.flink.api.table.Table
 
 import scala.language.reflectiveCalls
@@ -42,7 +44,10 @@ abstract class PlanTranslator {
   /**
    * Creates a [[Table]] from a DataSet (the underlying representation).
    */
-  def createTable[A](repr: Representation[A], fieldNames: Array[String]): Table
+  def createTable[A](
+    repr: Representation[A],
+    fieldIndexes: Array[Int],
+    fieldNames: Array[String]): Table
 
   /**
    * Creates a [[Table]] from the given DataSet.
@@ -50,10 +55,15 @@ abstract class PlanTranslator {
   def createTable[A](repr: Representation[A]): Table = {
 
     val fieldNames: Array[String] = repr.getType() match {
-      case c: CompositeType[A] => c.getFieldNames
-      case tpe => Array() // createTable will throw an exception for this later
+      case t: TupleTypeInfo[A] => t.getFieldNames
+      case c: CaseClassTypeInfo[A] => c.getFieldNames
+      case p: PojoTypeInfo[A] => p.getFieldNames
+      case tpe =>
+        throw new IllegalArgumentException(
+          s"Type $tpe requires explicit field naming with AS.")
     }
-    createTable(repr, fieldNames)
+    val fieldIndexes = fieldNames.indices.toArray
+    createTable(repr, fieldIndexes, fieldNames)
   }
 
   /**
@@ -75,17 +85,60 @@ abstract class PlanTranslator {
     */
   def createTable[A](repr: Representation[A], exprs: Array[Expression]): Table = {
 
-    val fieldNames: Array[String] = exprs
-      .map {
-        case ResolvedFieldReference(name, _) =>
-          name
-        case UnresolvedFieldReference(name) =>
-          name
-        case _ =>
-          throw new IllegalArgumentException("Only field expressions allowed")
-      }
-
-    createTable(repr, fieldNames)
+    val inputType = repr.getType()
+
+    val indexedNames: Array[(Int, String)] = inputType match {
+      case a: AtomicType[A] =>
+        if (exprs.length != 1) {
+          throw new IllegalArgumentException("Atomic type may can only have a single field.")
+        }
+        exprs.map {
+          case UnresolvedFieldReference(name) => (0, name)
+          case _ => throw new IllegalArgumentException(
+            "Field reference expression expected.")
+        }
+      case t: TupleTypeInfo[A] =>
+        exprs.zipWithIndex.map {
+          case (UnresolvedFieldReference(name), idx) => (idx, name)
+          case (Naming(UnresolvedFieldReference(origName), name), _) =>
+            val idx = t.getFieldIndex(origName)
+            if (idx < 0) {
+              throw new IllegalArgumentException(s"$origName is not a field of type $t")
+            }
+            (idx, name)
+          case _ => throw new IllegalArgumentException(
+            "Field reference expression or naming expression expected.")
+        }
+      case c: CaseClassTypeInfo[A] =>
+        exprs.zipWithIndex.map {
+          case (UnresolvedFieldReference(name), idx) => (idx, name)
+          case (Naming(UnresolvedFieldReference(origName), name), _) =>
+            val idx = c.getFieldIndex(origName)
+            if (idx < 0) {
+              throw new IllegalArgumentException(s"$origName is not a field of type $c")
+            }
+            (idx, name)
+          case _ => throw new IllegalArgumentException(
+            "Field reference expression or naming expression expected.")
+        }
+      case p: PojoTypeInfo[A] =>
+        exprs.map {
+          case Naming(UnresolvedFieldReference(origName), name) =>
+            val idx = p.getFieldIndex(origName)
+            if (idx < 0) {
+              throw new IllegalArgumentException(s"$origName is not a field of type $p")
+            }
+            (idx, name)
+          case _ => throw new IllegalArgumentException(
+            "Field naming expression expected.")
+        }
+      case tpe => throw new IllegalArgumentException(
+        s"Type $tpe cannot be converted into Table.")
+    }
+
+    val (fieldIndexes, fieldNames) = indexedNames.unzip
+
+    createTable(repr, fieldIndexes.toArray, fieldNames.toArray)
   }
 
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/1f3eda98/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 30a0589..f6fe2e4 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
@@ -22,19 +22,34 @@ import org.apache.calcite.sql.`type`.SqlTypeName
 import org.apache.calcite.sql.`type`.SqlTypeName._
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
 import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfo, GenericTypeInfo}
+import org.apache.flink.api.java.typeutils.ValueTypeInfo._
+import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
 
 object TypeConverter {
 
   def typeInfoToSqlType(typeInfo: TypeInformation[_]): SqlTypeName = typeInfo match {
     case BOOLEAN_TYPE_INFO => BOOLEAN
+    case BOOLEAN_VALUE_TYPE_INFO => BOOLEAN
     case BYTE_TYPE_INFO => TINYINT
+    case BYTE_VALUE_TYPE_INFO => TINYINT
     case SHORT_TYPE_INFO => SMALLINT
+    case SHORT_VALUE_TYPE_INFO => SMALLINT
     case INT_TYPE_INFO => INTEGER
+    case INT_VALUE_TYPE_INFO => INTEGER
     case LONG_TYPE_INFO => BIGINT
+    case LONG_VALUE_TYPE_INFO => BIGINT
     case FLOAT_TYPE_INFO => FLOAT
+    case FLOAT_VALUE_TYPE_INFO => FLOAT
     case DOUBLE_TYPE_INFO => DOUBLE
+    case DOUBLE_VALUE_TYPE_INFO => DOUBLE
     case STRING_TYPE_INFO => VARCHAR
+    case STRING_VALUE_TYPE_INFO => VARCHAR
     case DATE_TYPE_INFO => DATE
+//    case t: TupleTypeInfo[_] => ROW
+//    case c: CaseClassTypeInfo[_] => ROW
+//    case p: PojoTypeInfo[_] => STRUCTURED
+//    case g: GenericTypeInfo[_] => OTHER
     case _ => ??? // TODO more types
     }
 

http://git-wip-us.apache.org/repos/asf/flink/blob/1f3eda98/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
index effaf1a..53067dc 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
@@ -18,12 +18,25 @@
 
 package org.apache.flink.api.table.plan.nodes.dataset
 
+import java.lang.reflect.Field
+
 import org.apache.calcite.plan._
-import org.apache.calcite.rel.{RelWriter, RelNode}
+import org.apache.calcite.rel.RelNode
 import org.apache.calcite.rel.`type`.RelDataType
 import org.apache.calcite.rel.core.TableScan
+import org.apache.flink.api.common.functions.RichMapFunction
+import org.apache.flink.api.common.typeinfo.{TypeInformation, AtomicType}
 import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfo}
+import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
 import org.apache.flink.api.table.Row
+import org.apache.flink.api.table.plan.TypeConverter
+import org.apache.flink.api.table.plan.schema.DataSetTable
+import org.apache.flink.api.table.typeinfo.RowTypeInfo
+import org.apache.flink.configuration.Configuration
+
+import scala.collection.JavaConverters._
 
 /**
   * Flink RelNode which matches along with DataSource.
@@ -32,11 +45,12 @@ class DataSetSource(
     cluster: RelOptCluster,
     traitSet: RelTraitSet,
     table: RelOptTable,
-    rowType: RelDataType,
-    inputDataSet: DataSet[_])
+    rowType: RelDataType)
   extends TableScan(cluster, traitSet, table)
   with DataSetRel {
 
+  val dataSetTable: DataSetTable[Any] = table.unwrap(classOf[DataSetTable[Any]])
+
   override def deriveRowType() = rowType
 
   override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
@@ -44,12 +58,136 @@ class DataSetSource(
       cluster,
       traitSet,
       table,
-      rowType,
-      inputDataSet
+      rowType
     )
   }
 
   override def translateToPlan: DataSet[Any] = {
-    ???
+
+    val inputDataSet: DataSet[Any] = dataSetTable.dataSet
+
+    // extract Flink data types
+    val fieldTypes: Array[TypeInformation[_]] = rowType.getFieldList.asScala
+      .map(f => f.getType.getSqlTypeName)
+      .map(n => TypeConverter.sqlTypeToTypeInfo(n))
+      .toArray
+
+    val rowTypeInfo = new RowTypeInfo(fieldTypes, dataSetTable.fieldNames)
+
+    // convert input data set into row data set
+    inputDataSet.getType match {
+      case t: TupleTypeInfo[_] =>
+        val rowMapper = new TupleToRowMapper(dataSetTable.fieldIndexes)
+        inputDataSet.asInstanceOf[DataSet[Tuple]]
+          .map(rowMapper).returns(rowTypeInfo).asInstanceOf[DataSet[Any]]
+
+      case c: CaseClassTypeInfo[_] =>
+        val rowMapper = new CaseClassToRowMapper(dataSetTable.fieldIndexes)
+        inputDataSet.asInstanceOf[DataSet[Product]]
+          .map(rowMapper).returns(rowTypeInfo).asInstanceOf[DataSet[Any]]
+
+      case p: PojoTypeInfo[_] =>
+        // get pojo class
+        val typeClazz = p.getTypeClass.asInstanceOf[Class[Any]]
+        // get original field names
+        val origFieldNames = dataSetTable.fieldIndexes.map(i => p.getFieldNames()(i))
+
+        val rowMapper = new PojoToRowMapper(typeClazz, origFieldNames)
+        inputDataSet.asInstanceOf[DataSet[Any]]
+          .map(rowMapper).returns(rowTypeInfo).asInstanceOf[DataSet[Any]]
+
+      case a: AtomicType[_] =>
+        val rowMapper = new AtomicToRowMapper
+        inputDataSet.asInstanceOf[DataSet[Any]]
+          .map(rowMapper).returns(rowTypeInfo).asInstanceOf[DataSet[Any]]
+    }
+  }
+
+}
+
+class TupleToRowMapper(val fromIndexes: Array[Int])
+  extends RichMapFunction[Tuple, Row]
+{
+
+  @transient var outR: Row = null
+
+  override def open(conf: Configuration): Unit = {
+    outR = new Row(fromIndexes.length)
+  }
+
+  override def map(v: Tuple): Row = {
+
+    var i = 0
+    while (i < fromIndexes.length) {
+      outR.setField(i, v.getField(fromIndexes(i)))
+      i += 1
+    }
+    outR
+  }
+}
+
+class CaseClassToRowMapper(val fromIndexes: Array[Int])
+  extends RichMapFunction[Product, Row]
+{
+
+  @transient var outR: Row = null
+
+  override def open(conf: Configuration): Unit = {
+    outR = new Row(fromIndexes.length)
+  }
+
+  override def map(v: Product): Row = {
+
+    var i = 0
+    while (i < fromIndexes.length) {
+      outR.setField(i, v.productElement(fromIndexes(i)))
+      i += 1
+    }
+    outR
+  }
+}
+
+class PojoToRowMapper(val inClazz: Class[Any], val fieldNames: Array[String])
+  extends RichMapFunction[Any, Row]
+{
+
+  @transient var outR: Row = null
+  @transient var fields: Array[Field] = null
+
+  override def open(conf: Configuration): Unit = {
+
+    fields = fieldNames.map { n =>
+      val f = inClazz.getField(n)
+      f.setAccessible(true)
+      f
+    }
+    outR = new Row(fieldNames.length)
+  }
+
+  override def map(v: Any): Row = {
+
+    var i = 0
+    while (i < fields.length) {
+      outR.setField(i, fields(i).get(v))
+      i += 1
+    }
+    outR
+  }
+}
+
+class AtomicToRowMapper()
+  extends RichMapFunction[Any, Row]
+{
+
+  @transient var outR: Row = null
+
+  override def open(conf: Configuration): Unit = {
+    outR = new Row(1)
+  }
+
+  override def map(v: Any): Row = {
+
+    outR.setField(0, v)
+    outR
   }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/1f3eda98/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
index 937f3e2..f995201 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
@@ -21,10 +21,8 @@ package org.apache.flink.api.table.plan.rules.dataset
 import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
 import org.apache.calcite.rel.RelNode
 import org.apache.calcite.rel.convert.ConverterRule
-import org.apache.flink.api.java.DataSet
 import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetSource}
 import org.apache.flink.api.table.plan.nodes.logical.{FlinkScan, FlinkConvention}
-import org.apache.flink.api.table.plan.schema.DataSetTable
 
 class DataSetScanRule
   extends ConverterRule(
@@ -33,17 +31,17 @@ class DataSetScanRule
     DataSetConvention.INSTANCE,
     "DataSetScanRule")
 {
+
   def convert(rel: RelNode): RelNode = {
     val scan: FlinkScan = rel.asInstanceOf[FlinkScan]
     val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
-    val dataSet: DataSet[_] = scan.getTable().unwrap(classOf[DataSetTable[_]]).dataSet
+
 
     new DataSetSource(
       rel.getCluster,
       traitSet,
       scan.getTable,
-      rel.getRowType,
-      dataSet
+      rel.getRowType
     )
   }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/1f3eda98/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala
index e6aecab..75090a2 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala
@@ -28,40 +28,45 @@ import org.apache.calcite.schema.Statistic
 import org.apache.calcite.schema.impl.AbstractTable
 import org.apache.calcite.sql.`type`.SqlTypeName
 import org.apache.calcite.util.ImmutableBitSet
+import org.apache.flink.api.common.typeinfo.AtomicType
 import org.apache.flink.api.common.typeutils.CompositeType
 import org.apache.flink.api.java.DataSet
 import org.apache.flink.api.table.plan.TypeConverter
 
 class DataSetTable[T](
     val dataSet: DataSet[T],
+    val fieldIndexes: Array[Int],
     val fieldNames: Array[String])
   extends AbstractTable {
 
+  if (fieldIndexes.length != fieldNames.length) {
+    throw new IllegalArgumentException(
+      "Number of field indexes and field names must be equal.")
+  }
+
   // check uniquenss of field names
   if (fieldNames.length != fieldNames.toSet.size) {
-    throw new scala.IllegalArgumentException(
+    throw new IllegalArgumentException(
       "Table field names must be unique.")
   }
 
-  val dataSetType: CompositeType[T] =
+  val fieldTypes: Array[SqlTypeName] =
     dataSet.getType match {
       case cType: CompositeType[T] =>
-        cType
-      case _ =>
-        throw new scala.IllegalArgumentException(
-          "DataSet must have a composite type.")
-    }
-
-  val fieldTypes: Array[SqlTypeName] =
-    if (fieldNames.length == dataSetType.getArity) {
-      (0 until dataSetType.getArity)
-        .map(i => dataSetType.getTypeAt(i))
-        .map(TypeConverter.typeInfoToSqlType)
-        .toArray
-    }
-    else {
-      throw new IllegalArgumentException(
-        "Arity of DataSet type not equal to number of field names.")
+        if (fieldNames.length != cType.getArity) {
+          throw new IllegalArgumentException(
+          s"Arity of DataSet type (" + cType.getFieldNames.deep + ") " +
+            "not equal to number of field names " + fieldNames.deep + ".")
+        }
+        fieldIndexes
+          .map(cType.getTypeAt(_))
+          .map(TypeConverter.typeInfoToSqlType(_))
+      case aType: AtomicType[T] =>
+        if (fieldIndexes.length != 1 || fieldIndexes(0) != 0) {
+          throw new IllegalArgumentException(
+            "Non-composite input type may have only a single field and its index must be
0.")
+        }
+        Array(TypeConverter.typeInfoToSqlType(aType))
     }
 
   override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {

http://git-wip-us.apache.org/repos/asf/flink/blob/1f3eda98/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 f257c32..0f35d75 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
@@ -28,8 +28,8 @@ 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.ArrayList;
 import java.util.List;
 
 @RunWith(Parameterized.class)
@@ -40,8 +40,8 @@ public class AsITCase extends MultipleProgramsTestBase {
 		super(mode);
 	}
 
-	@Test(expected = NotImplementedError.class)
-	public void testAs() throws Exception {
+	@Test
+	public void testAsFromTuple() throws Exception {
 		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
 		TableEnvironment tableEnv = new TableEnvironment();
 
@@ -50,12 +50,39 @@ public class AsITCase extends MultipleProgramsTestBase {
 
 		DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
 		List<Row> results = ds.collect();
-		String 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" + "7,4," +
-				"Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5,"
+
-				"Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5,"
+
-				"Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19,"
+
-				"6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
+		String 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" +
+			"7,4,Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" +
+			"11,5,Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" +
+			"14,5,Comment#8\n" + "15,5,Comment#9\n" + "16,6,Comment#10\n" +
+			"17,6,Comment#11\n" + "18,6,Comment#12\n" + "19,6,Comment#13\n" +
+			"20,6,Comment#14\n" + "21,6,Comment#15\n";
+		compareResultAsText(results, expected);
+	}
+
+	@Test
+	public void testAsFromPojo() throws Exception {
+		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		TableEnvironment tableEnv = new TableEnvironment();
+
+		List<SmallPojo> data = new ArrayList<>();
+		data.add(new SmallPojo("Peter", 28, 4000.00, "Sales"));
+		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),
+				"department AS a, " +
+				"age AS b, " +
+				"salary AS c, " +
+				"name AS d");
+
+		DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+		List<Row> results = ds.collect();
+		String expected =
+			"Sales,28,4000.0,Peter\n" +
+			"Engineering,56,10000.0,Anna\n" +
+			"HR,42,6000.0,Lucy\n";
 		compareResultAsText(results, expected);
 	}
 
@@ -129,5 +156,23 @@ public class AsITCase extends MultipleProgramsTestBase {
 		String expected = "";
 		compareResultAsText(results, expected);
 	}
+
+	public static class SmallPojo {
+
+		public SmallPojo() { }
+
+		public SmallPojo(String name, int age, double salary, String department) {
+			this.name = name;
+			this.age = age;
+			this.salary = salary;
+			this.department = department;
+		}
+
+		public String name;
+		public int age;
+		public double salary;
+		public String department;
+	}
+
 }
 

http://git-wip-us.apache.org/repos/asf/flink/blob/1f3eda98/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
index a3ab10f..cd08879 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
@@ -60,7 +60,7 @@ public class FilterITCase extends MultipleProgramsTestBase {
 		compareResultAsText(results, expected);
 	}
 
-	@Test(expected = NotImplementedError.class)
+	@Test
 	public void testAllPassingFilter() throws Exception {
 		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
 		TableEnvironment tableEnv = new TableEnvironment();
@@ -75,12 +75,13 @@ public class FilterITCase extends MultipleProgramsTestBase {
 
 		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		List<Row> results = ds.collect();
-		String 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" + "7,4," +
-				"Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5,"
+
-				"Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5,"
+
-				"Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19,"
+
-				"6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
+		String 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" +
+			"7,4,Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" +
+			"11,5,Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" +
+			"14,5,Comment#8\n" + "15,5,Comment#9\n" + "16,6,Comment#10\n" +
+			"17,6,Comment#11\n" + "18,6,Comment#12\n" + "19,6,Comment#13\n" +
+			"20,6,Comment#14\n" + "21,6,Comment#15\n";
 		compareResultAsText(results, expected);
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/1f3eda98/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
index a3d31da..a66219c 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
@@ -41,7 +41,7 @@ public class SelectITCase extends MultipleProgramsTestBase {
 		super(mode);
 	}
 
-	@Test(expected = NotImplementedError.class)
+	@Test
 	public void testSimpleSelectAllWithAs() throws Exception {
 		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
 		TableEnvironment tableEnv = new TableEnvironment();
@@ -55,12 +55,13 @@ public class SelectITCase extends MultipleProgramsTestBase {
 
 		DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
 		List<Row> results = resultSet.collect();
-		String 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" + "7,4," +
-				"Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5,"
+
-				"Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5,"
+
-				"Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19,"
+
-				"6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
+		String 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" +
+			"7,4,Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" +
+			"11,5,Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" +
+			"14,5,Comment#8\n" + "15,5,Comment#9\n" + "16,6,Comment#10\n" +
+			"17,6,Comment#11\n" + "18,6,Comment#12\n" + "19,6,Comment#13\n" +
+			"20,6,Comment#14\n" + "21,6,Comment#15\n";
 		compareResultAsText(results, expected);
 
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/1f3eda98/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 5ff2b82..6779d4c 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
@@ -33,18 +33,38 @@ import scala.collection.JavaConverters._
 @RunWith(classOf[Parameterized])
 class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
 
-  @Test(expected = classOf[NotImplementedError])
+  @Test
   def testAs(): Unit = {
 
     val env = ExecutionEnvironment.getExecutionEnvironment
     val t = CollectionDataSets.get3TupleDataSet(env).as('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" + "7,4," +
-      "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5,"
+
-      "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5,"
+
-      "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19,"
+
-      "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
+    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" +
+      "7,4,Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" +
+      "11,5,Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" +
+      "15,5,Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n"
+
+      "19,6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
+    val results = t.toDataSet[Row].collect()
+    TestBaseUtils.compareResultAsText(results.asJava, expected)
+  }
+
+  @Test
+  @throws(classOf[Exception])
+  def testAsFromCaseClass(): Unit = {
+
+    val env = ExecutionEnvironment.getExecutionEnvironment
+    val data = List(
+      SomeCaseClass("Peter", 28, 4000.00, "Sales"),
+      SomeCaseClass("Anna", 56, 10000.00, "Engineering"),
+      SomeCaseClass("Lucy", 42, 6000.00, "HR"))
+
+    val t =  env.fromCollection(data).as('a, 'b, 'c, 'd)
+
+    val expected: String =
+      "Peter,28,4000.0,Sales\n" +
+      "Anna,56,10000.0,Engineering\n" +
+      "Lucy,42,6000.0,HR\n"
     val results = t.toDataSet[Row].collect()
     TestBaseUtils.compareResultAsText(results.asJava, expected)
   }
@@ -105,4 +125,10 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode)
{
     val results = t.toDataSet[Row].collect()
     TestBaseUtils.compareResultAsText(results.asJava, expected)
   }
+
+}
+
+case class SomeCaseClass(name: String, age: Int, salary: Double, department: String) {
+
+  def this() { this("", 0, 0.0, "") }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/1f3eda98/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala
index 4dadfe4..3700d67 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala
@@ -33,34 +33,34 @@ import scala.collection.JavaConverters._
 @RunWith(classOf[Parameterized])
 class SelectITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
 
-  @Test(expected = classOf[NotImplementedError])
+  @Test
   def testSimpleSelectAll(): Unit = {
 
     val env = ExecutionEnvironment.getExecutionEnvironment
     val t = CollectionDataSets.get3TupleDataSet(env).toTable.select('_1, '_2, '_3)
 
-    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" + "7,4," +
-      "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5,"
+
-      "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5,"
+
-      "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19,"
+
-      "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
+    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" +
+      "7,4,Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" +
+      "11,5,Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" +
+      "15,5,Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n"
+
+      "19,6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
     val results = t.toDataSet[Row].collect()
     TestBaseUtils.compareResultAsText(results.asJava, expected)
   }
 
-  @Test(expected = classOf[NotImplementedError])
+  @Test
   def testSimpleSelectAllWithAs(): Unit = {
 
     val env = ExecutionEnvironment.getExecutionEnvironment
     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" + "7,4," +
-      "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5,"
+
-      "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5,"
+
-      "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19,"
+
-      "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
+    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" +
+      "7,4,Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" +
+      "11,5,Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" +
+      "15,5,Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n"
+
+      "19,6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
     val results = t.toDataSet[Row].collect()
     TestBaseUtils.compareResultAsText(results.asJava, expected)
   }


Mime
View raw message