From 4def008717cec7ad7a97e9e24f572e90ef21cba6 Mon Sep 17 00:00:00 2001 From: twalthr Date: Thu, 11 Feb 2016 16:16:29 +0100 Subject: [PATCH 1/2] [FLINK-3226] Translation from and to POJOs for CodeGenerator --- .../api/java/table/TableEnvironment.scala | 10 + .../api/table/codegen/CodeGenUtils.scala | 59 +++- .../api/table/codegen/CodeGenerator.scala | 251 ++++++++++++---- .../flink/api/table/plan/TypeConverter.scala | 36 ++- .../plan/nodes/dataset/DataSetSource.scala | 25 +- .../rules/dataset/DataSetFilterRule.scala | 7 +- .../rules/dataset/DataSetProjectRule.scala | 5 +- .../flink/api/java/table/test/AsITCase.java | 277 +++++++++++++++++- .../flink/api/scala/table/test/AsITCase.scala | 45 ++- 9 files changed, 616 insertions(+), 99 deletions(-) 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 2027037bafc9a1..edb3723e8b6d1b 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 @@ -75,5 +75,15 @@ class TableEnvironment { TypeExtractor.createTypeInfo(clazz).asInstanceOf[TypeInformation[T]]) } + /** + * 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 + * fields and the types must match. + */ + def toDataSet[T](table: Table, typeInfo: TypeInformation[T]): DataSet[T] = { + new JavaBatchTranslator(config).translate[T](table.relNode)(typeInfo) + } + } diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala index 5bd14679a96bc9..fd55cd84abb469 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala @@ -18,13 +18,14 @@ package org.apache.flink.api.table.codegen +import java.lang.reflect.Field import java.util.concurrent.atomic.AtomicInteger import org.apache.flink.api.common.typeinfo.BasicTypeInfo._ import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo._ import org.apache.flink.api.common.typeinfo.{NumericTypeInfo, TypeInformation} import org.apache.flink.api.common.typeutils.CompositeType -import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfo} +import org.apache.flink.api.java.typeutils.{TypeExtractor, PojoTypeInfo, TupleTypeInfo} import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo import org.apache.flink.api.table.typeinfo.RowTypeInfo @@ -150,7 +151,11 @@ object CodeGenUtils { sealed abstract class FieldAccessor - case class ObjectFieldAccessor(fieldName: String) extends FieldAccessor + case class ObjectFieldAccessor(field: Field) extends FieldAccessor + + case class ObjectGenericFieldAccessor(name: String) extends FieldAccessor + + case class ObjectPrivateFieldAccessor(field: Field) extends FieldAccessor case class ObjectMethodAccessor(methodName: String) extends FieldAccessor @@ -165,12 +170,56 @@ object CodeGenUtils { ObjectMethodAccessor(cc.getFieldNames()(index)) case javaTup: TupleTypeInfo[_] => - ObjectFieldAccessor("f" + index) + ObjectGenericFieldAccessor("f" + index) - case pj: PojoTypeInfo[_] => - ObjectFieldAccessor(pj.getFieldNames()(index)) + case pt: PojoTypeInfo[_] => + val fieldName = pt.getFieldNames()(index) + getFieldAccessor(pt.getTypeClass, fieldName) case _ => throw new CodeGenException("Unsupported composite type.") } } + + def getFieldAccessor(clazz: Class[_], fieldName: String): FieldAccessor = { + val field = TypeExtractor.getDeclaredField(clazz, fieldName) + if (field.isAccessible) { + ObjectFieldAccessor(field) + } + else { + ObjectPrivateFieldAccessor(field) + } + } + + def isFieldPrimitive(field: Field): Boolean = field.getType.isPrimitive + + def reflectiveFieldReadAccess(fieldTerm: String, field: Field, objectTerm: String): String = + field.getType match { + case java.lang.Integer.TYPE => s"$fieldTerm.getInt($objectTerm)" + case java.lang.Long.TYPE => s"$fieldTerm.getLong($objectTerm)" + case java.lang.Short.TYPE => s"$fieldTerm.getShort($objectTerm)" + case java.lang.Byte.TYPE => s"$fieldTerm.getByte($objectTerm)" + case java.lang.Float.TYPE => s"$fieldTerm.getFloat($objectTerm)" + case java.lang.Double.TYPE => s"$fieldTerm.getDouble($objectTerm)" + case java.lang.Boolean.TYPE => s"$fieldTerm.getBoolean($objectTerm)" + case java.lang.Character.TYPE => s"$fieldTerm.getChar($objectTerm)" + case _ => s"(${field.getType.getCanonicalName}) $fieldTerm.get($objectTerm)" + } + + def reflectiveFieldWriteAccess( + fieldTerm: String, + field: Field, + objectTerm: String, + valueTerm: String) + : String = + field.getType match { + case java.lang.Integer.TYPE => s"$fieldTerm.setInt($objectTerm, $valueTerm)" + case java.lang.Long.TYPE => s"$fieldTerm.setLong($objectTerm, $valueTerm)" + case java.lang.Short.TYPE => s"$fieldTerm.setShort($objectTerm, $valueTerm)" + case java.lang.Byte.TYPE => s"$fieldTerm.setByte($objectTerm, $valueTerm)" + case java.lang.Float.TYPE => s"$fieldTerm.setFloat($objectTerm, $valueTerm)" + case java.lang.Double.TYPE => s"$fieldTerm.setDouble($objectTerm, $valueTerm)" + case java.lang.Boolean.TYPE => s"$fieldTerm.setBoolean($objectTerm, $valueTerm)" + case java.lang.Character.TYPE => s"$fieldTerm.setChar($objectTerm, $valueTerm)" + case _ => s"$fieldTerm.set($objectTerm, $valueTerm)" + } } diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala index a4ae4b1407a296..1f71b622618576 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala @@ -42,13 +42,30 @@ import scala.collection.mutable * @param config configuration that determines runtime behavior * @param input1 type information about the first input of the Function * @param input2 type information about the second input if the Function is binary + * @param inputPojoFieldMapping additional mapping information if input1 is a POJO (POJO types + * have no deterministic field order). We assume that input2 is + * converted before and thus is never a POJO. */ class CodeGenerator( - config: TableConfig, - input1: TypeInformation[Any], - input2: Option[TypeInformation[Any]] = None) + config: TableConfig, + input1: TypeInformation[Any], + input2: Option[TypeInformation[Any]] = None, + inputPojoFieldMapping: Array[Int] = Array()) extends RexVisitor[GeneratedExpression] { + /** + * A code generator for generating unary Flink + * [[org.apache.flink.api.common.functions.Function]]s with one input. + * + * @param config configuration that determines runtime behavior + * @param input type information about the input of the Function + * @param inputPojoFieldMapping additional mapping information necessary if input is a + * POJO (POJO types have no deterministic field order). + */ + def this(config: TableConfig, input: TypeInformation[Any], inputPojoFieldMapping: Array[Int]) = + this(config, input, None, inputPojoFieldMapping) + + // set of member statements that will be added only once // we use a LinkedHashSet to keep the insertion order private val reusableMemberStatements = mutable.LinkedHashSet[String]() @@ -169,12 +186,12 @@ class CodeGenerator( ${reuseMemberCode()} - public $funcName() { + public $funcName() throws Exception{ ${reuseInitCode()} } @Override - public ${samHeader._1} { + public ${samHeader._1} throws Exception { ${samHeader._2} ${reuseInputUnboxingCode()} $bodyCode @@ -192,10 +209,12 @@ class CodeGenerator( * may be stored in the global result variable (see [[outRecordTerm]]). * * @param returnType conversion target type. Inputs and output must have the same arity. + * @param resultFieldNames result field names necessary for a mapping to POJO fields. * @return instance of GeneratedExpression */ def generateConverterResultExpression( - returnType: TypeInformation[_ <: Any]) + returnType: TypeInformation[_ <: Any], + resultFieldNames: Seq[String]) : GeneratedExpression = { val input1AccessExprs = for (i <- 0 until input1.getArity) yield generateInputAccess(input1, input1Term, i) @@ -206,7 +225,7 @@ class CodeGenerator( case None => Seq() // add nothing } - generateResultExpression(input1AccessExprs ++ input2AccessExprs, returnType) + generateResultExpression(input1AccessExprs ++ input2AccessExprs, returnType, resultFieldNames) } /** @@ -215,15 +234,17 @@ class CodeGenerator( * may be stored in the global result variable (see [[outRecordTerm]]). * * @param returnType conversion target type. Type must have the same arity than rexNodes. - * @param rexNodes sequence of RexNode + * @param resultFieldNames result field names necessary for a mapping to POJO fields. + * @param rexNodes sequence of RexNode to be converted * @return instance of GeneratedExpression */ def generateResultExpression( returnType: TypeInformation[_ <: Any], + resultFieldNames: Seq[String], rexNodes: Seq[RexNode]) : GeneratedExpression = { val fieldExprs = rexNodes.map(generateExpression) - generateResultExpression(fieldExprs, returnType) + generateResultExpression(fieldExprs, returnType, resultFieldNames) } /** @@ -231,29 +252,40 @@ class CodeGenerator( * be reused, they will be added to reusable code sections internally. The evaluation result * may be stored in the global result variable (see [[outRecordTerm]]). * - * @param fieldExprs + * @param fieldExprs field expressions to be converted * @param returnType conversion target type. Type must have the same arity than fieldExprs. + * @param resultFieldNames result field names necessary for a mapping to POJO fields. * @return instance of GeneratedExpression */ def generateResultExpression( fieldExprs: Seq[GeneratedExpression], - returnType: TypeInformation[_ <: Any]) + returnType: TypeInformation[_ <: Any], + resultFieldNames: Seq[String]) : GeneratedExpression = { - // TODO disable arity check for Rows and derive row arity from fieldExprs // initial type check if (returnType.getArity != fieldExprs.length) { throw new CodeGenException("Arity of result type does not match number of expressions.") } // type check returnType match { + case pt: PojoTypeInfo[_] => + fieldExprs.zipWithIndex foreach { + case (fieldExpr, i) if fieldExpr.resultType != pt.getTypeAt(resultFieldNames(i)) => + throw new CodeGenException("Incompatible types of expression and result type.") + + case _ => // ok + } + case ct: CompositeType[_] => fieldExprs.zipWithIndex foreach { case (fieldExpr, i) if fieldExpr.resultType != ct.getTypeAt(i) => throw new CodeGenException("Incompatible types of expression and result type.") case _ => // ok } + case at: AtomicType[_] if at != fieldExprs.head.resultType => throw new CodeGenException("Incompatible types of expression and result type.") + case _ => // ok } @@ -286,28 +318,74 @@ class CodeGenerator( GeneratedExpression(outRecordTerm, "false", resultSetters, returnType) - case pj: PojoTypeInfo[_] => - addReusableOutRecord(pj) - val resultSetters: String = fieldExprs.zip(pj.getFieldNames) map { - case (fieldExpr, fieldName) => - if (nullCheck) { - s""" - |${fieldExpr.code} - |if (${fieldExpr.nullTerm}) { - | $outRecordTerm.$fieldName = null; - |} - |else { - | $outRecordTerm.$fieldName = ${fieldExpr.resultTerm}; - |} - |""".stripMargin - } - else { - s""" - |${fieldExpr.code} - |$outRecordTerm.$fieldName = ${fieldExpr.resultTerm}; - |""".stripMargin - } - } mkString "\n" + case pt: PojoTypeInfo[_] => + addReusableOutRecord(pt) + val resultSetters: String = fieldExprs.zip(resultFieldNames) map { + case (fieldExpr, fieldName) => + val accessor = getFieldAccessor(pt.getTypeClass, fieldName) + + accessor match { + // Reflective access of primitives/Objects + case ObjectPrivateFieldAccessor(field) => + val fieldTerm = addReusablePrivateFieldAccess(pt.getTypeClass, fieldName) + + val defaultIfNull = if (isFieldPrimitive(field)) { + primitiveDefaultValue(fieldExpr.resultType) + } else { + "null" + } + + if (nullCheck) { + s""" + |${fieldExpr.code} + |if (${fieldExpr.nullTerm}) { + | ${reflectiveFieldWriteAccess( + fieldTerm, + field, + outRecordTerm, + defaultIfNull)}; + |} + |else { + | ${reflectiveFieldWriteAccess( + fieldTerm, + field, + outRecordTerm, + fieldExpr.resultTerm)}; + |} + |""".stripMargin + } + else { + s""" + |${fieldExpr.code} + |${reflectiveFieldWriteAccess( + fieldTerm, + field, + outRecordTerm, + fieldExpr.resultTerm)}; + |""".stripMargin + } + + // primitive or Object field access (implicit boxing) + case _ => + if (nullCheck) { + s""" + |${fieldExpr.code} + |if (${fieldExpr.nullTerm}) { + | $outRecordTerm.$fieldName = null; + |} + |else { + | $outRecordTerm.$fieldName = ${fieldExpr.resultTerm}; + |} + |""".stripMargin + } + else { + s""" + |${fieldExpr.code} + |$outRecordTerm.$fieldName = ${fieldExpr.resultTerm}; + |""".stripMargin + } + } + } mkString "\n" GeneratedExpression(outRecordTerm, "false", resultSetters, returnType) @@ -619,29 +697,67 @@ class CodeGenerator( case Some(expr) => expr - // generate input access and boxing + // generate input access and boxing if necessary case None => val newExpr = inputType match { + case ct: CompositeType[_] => - val accessor = fieldAccessorFor(ct, index) - val fieldType: TypeInformation[Any] = ct.getTypeAt(index) + val fieldIndex = if (ct.isInstanceOf[PojoTypeInfo[_]]) { + inputPojoFieldMapping(index) + } + else { + index + } + val accessor = fieldAccessorFor(ct, fieldIndex) + val fieldType: TypeInformation[Any] = ct.getTypeAt(fieldIndex) val fieldTypeTerm = boxedTypeTermForTypeInfo(fieldType) - val inputCode = accessor match { - case ObjectFieldAccessor(fieldName) => - s"($fieldTypeTerm) $inputTerm.$fieldName" + accessor match { + case ObjectFieldAccessor(field) => + // primitive + if (isFieldPrimitive(field)) { + generateNonNullLiteral(fieldType, s"$inputTerm.${field.getName}") + } + // Object + else { + generateNullableLiteral( + fieldType, + s"($fieldTypeTerm) $inputTerm.${field.getName}") + } + + case ObjectGenericFieldAccessor(fieldName) => + // Object + val inputCode = s"($fieldTypeTerm) $inputTerm.$fieldName" + generateNullableLiteral(fieldType, inputCode) case ObjectMethodAccessor(methodName) => - s"($fieldTypeTerm) $inputTerm.$methodName()" + // Object + val inputCode = s"($fieldTypeTerm) $inputTerm.$methodName()" + generateNullableLiteral(fieldType, inputCode) case ProductAccessor(i) => - s"($fieldTypeTerm) $inputTerm.productElement($i)" + // Object + val inputCode = s"($fieldTypeTerm) $inputTerm.productElement($i)" + generateNullableLiteral(fieldType, inputCode) + + case ObjectPrivateFieldAccessor(field) => + val fieldTerm = addReusablePrivateFieldAccess(ct.getTypeClass, field.getName) + val reflectiveAccessCode = reflectiveFieldReadAccess(fieldTerm, field, inputTerm) + // primitive + if (isFieldPrimitive(field)) { + generateNonNullLiteral(fieldType, reflectiveAccessCode) + } + // Object + else { + generateNullableLiteral(fieldType, reflectiveAccessCode) + } } - generateInputUnboxing(fieldType, inputCode) + case at: AtomicType[_] => val fieldTypeTerm = boxedTypeTermForTypeInfo(at) val inputCode = s"($fieldTypeTerm) $inputTerm" - generateInputUnboxing(at, inputCode) + generateNullableLiteral(at, inputCode) + case _ => throw new CodeGenException("Unsupported type for input access.") } @@ -652,20 +768,20 @@ class CodeGenerator( GeneratedExpression(inputExpr.resultTerm, inputExpr.nullTerm, "", inputExpr.resultType) } - private def generateInputUnboxing( - inputType: TypeInformation[Any], - inputCode: String) + private def generateNullableLiteral( + literalType: TypeInformation[Any], + literalCode: String) : GeneratedExpression = { val tmpTerm = newName("tmp") val resultTerm = newName("result") val nullTerm = newName("isNull") - val tmpTypeTerm = boxedTypeTermForTypeInfo(inputType) - val resultTypeTerm = primitiveTypeTermForTypeInfo(inputType) - val defaultValue = primitiveDefaultValue(inputType) + val tmpTypeTerm = boxedTypeTermForTypeInfo(literalType) + val resultTypeTerm = primitiveTypeTermForTypeInfo(literalType) + val defaultValue = primitiveDefaultValue(literalType) - val wrappedCode = if (nullCheck && !isReference(inputType)) { + val wrappedCode = if (nullCheck && !isReference(literalType)) { s""" - |$tmpTypeTerm $tmpTerm = $inputCode; + |$tmpTypeTerm $tmpTerm = $literalCode; |boolean $nullTerm = $tmpTerm == null; |$resultTypeTerm $resultTerm; |if ($nullTerm) { @@ -677,16 +793,16 @@ class CodeGenerator( |""".stripMargin } else if (nullCheck) { s""" - |$resultTypeTerm $resultTerm = $inputCode; - |boolean $nullTerm = $inputCode == null; + |$resultTypeTerm $resultTerm = $literalCode; + |boolean $nullTerm = $literalCode == null; |""".stripMargin } else { s""" - |$resultTypeTerm $resultTerm = $inputCode; + |$resultTypeTerm $resultTerm = $literalCode; |""".stripMargin } - GeneratedExpression(resultTerm, nullTerm, wrappedCode, inputType) + GeneratedExpression(resultTerm, nullTerm, wrappedCode, literalType) } private def generateNonNullLiteral( @@ -731,9 +847,11 @@ class CodeGenerator( GeneratedExpression(resultTerm, nullTerm, wrappedCode, resultType) } + // ---------------------------------------------------------------------------------------------- + // Reusable code snippets // ---------------------------------------------------------------------------------------------- - def addReusableOutRecord(ti: TypeInformation[_]) = { + def addReusableOutRecord(ti: TypeInformation[_]): Unit = { val statement = ti match { case rt: RowTypeInfo => s""" @@ -749,4 +867,23 @@ class CodeGenerator( reusableMemberStatements.add(statement) } + def addReusablePrivateFieldAccess(clazz: Class[_], fieldName: String): String = { + val fieldTerm = s"field_${clazz.getCanonicalName.replace('.', '$')}_$fieldName" + val fieldExtraction = + s""" + |java.lang.reflect.Field $fieldTerm = + | org.apache.flink.api.java.typeutils.TypeExtractor.getDeclaredField( + | ${clazz.getCanonicalName}.class, "$fieldName"); + |""".stripMargin + reusableMemberStatements.add(fieldExtraction) + + val fieldAccessibility = + s""" + |$fieldTerm.setAccessible(true); + |""".stripMargin + reusableInitStatements.add(fieldAccessibility) + + fieldTerm + } + } 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 1fc482ae873cfc..9a96bf52e9e5d8 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 @@ -19,21 +19,21 @@ package org.apache.flink.api.table.plan import org.apache.calcite.rel.`type`.RelDataType +import org.apache.calcite.rel.core.JoinRelType import org.apache.calcite.rel.core.JoinRelType._ +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.{AtomicType, TypeInformation} import org.apache.flink.api.common.typeutils.CompositeType +import org.apache.flink.api.java.operators.join.JoinType import org.apache.flink.api.java.tuple.Tuple -import org.apache.flink.api.java.typeutils.TupleTypeInfo import org.apache.flink.api.java.typeutils.ValueTypeInfo._ +import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfo} import org.apache.flink.api.table.typeinfo.RowTypeInfo import org.apache.flink.api.table.{Row, TableException} + import scala.collection.JavaConversions._ -import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo -import org.apache.flink.api.java.operators.join.JoinType -import org.apache.calcite.rel.core.JoinRelType -import org.apache.calcite.sql.`type`.SqlTypeName object TypeConverter { @@ -87,6 +87,8 @@ object TypeConverter { val logicalFieldTypes = logicalRowType.getFieldList map { relDataType => TypeConverter.sqlTypeToTypeInfo(relDataType.getType.getSqlTypeName) } + // field names + val logicalFieldNames = logicalRowType.getFieldNames val returnType = expectedPhysicalType match { // a certain physical type is expected (but not Row) @@ -96,19 +98,39 @@ object TypeConverter { throw new TableException("Arity of result does not match expected type.") } typeInfo match { + + // 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") + } + val expectedTypeInfo = pt.getTypeAt(i) + if (fieldTypeInfo != expectedTypeInfo) { + throw new TableException(s"Result field does not match expected type. " + + s"Expected: $expectedTypeInfo; Actual: $fieldTypeInfo") + } + } + + // Tuple/Case class type expected case ct: CompositeType[_] => logicalFieldTypes.zipWithIndex foreach { case (fieldTypeInfo, i) => val expectedTypeInfo = ct.getTypeAt(i) if (fieldTypeInfo != expectedTypeInfo) { - throw new TableException(s"Result field does not match expected type." + + throw new TableException(s"Result field does not match expected type. " + s"Expected: $expectedTypeInfo; Actual: $fieldTypeInfo") } } + + // Atomic type expected case at: AtomicType[_] => val fieldTypeInfo = logicalFieldTypes.head if (fieldTypeInfo != at) { - throw new TableException(s"Result field does not match expected type." + + throw new TableException(s"Result field does not match expected type. " + s"Expected: $at; Actual: $fieldTypeInfo") } 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 33fe43051c2e0b..50e01e7c3a43f5 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 @@ -25,14 +25,19 @@ import org.apache.calcite.rel.core.TableScan import org.apache.flink.api.common.functions.MapFunction import org.apache.flink.api.common.typeinfo.TypeInformation import org.apache.flink.api.java.DataSet +import org.apache.flink.api.java.typeutils.PojoTypeInfo import org.apache.flink.api.table.TableConfig import org.apache.flink.api.table.codegen.CodeGenerator import org.apache.flink.api.table.plan.TypeConverter.determineReturnType import org.apache.flink.api.table.plan.schema.DataSetTable import org.apache.flink.api.table.runtime.MapRunner +import scala.collection.JavaConversions._ + /** * Flink RelNode which matches along with DataSource. + * It ensures that types without deterministic field order (e.g. POJOs) are not part of + * the plan translation. */ class DataSetSource( cluster: RelOptCluster, @@ -63,11 +68,13 @@ class DataSetSource( val inputDataSet: DataSet[Any] = dataSetTable.dataSet val inputType = inputDataSet.getType - // special case: - // if efficient type usage is enabled and no expected type is set - // we can simply forward the DataSet to the next operator expectedType match { - case None if config.getEfficientTypeUsage => + + // special case: + // if efficient type usage is enabled and no expected type is set + // we can simply forward the DataSet to the next operator. + // however, we cannot forward PojoTypes as their fields don't have an order + case None if config.getEfficientTypeUsage && !inputType.isInstanceOf[PojoTypeInfo[_]] => inputDataSet case _ => @@ -79,8 +86,14 @@ class DataSetSource( // conversion if (determinedType != inputType) { - val generator = new CodeGenerator(config, inputDataSet.getType) - val conversion = generator.generateConverterResultExpression(determinedType) + val generator = new CodeGenerator( + config, + inputDataSet.getType, + dataSetTable.fieldIndexes) + + val conversion = generator.generateConverterResultExpression( + determinedType, + getRowType.getFieldNames) val body = s""" diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala index 0ca153d89725b0..bec4d1f9d07891 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala @@ -29,6 +29,8 @@ import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSet import org.apache.flink.api.table.plan.nodes.logical.{FlinkConvention, FlinkFilter} import org.apache.flink.api.table.runtime.FlatMapRunner +import scala.collection.JavaConversions._ + class DataSetFilterRule extends ConverterRule( classOf[FlinkFilter], @@ -52,7 +54,10 @@ class DataSetFilterRule // conversion val body = if (inputType != returnType) { - val conversion = generator.generateConverterResultExpression(returnType) + val conversion = generator.generateConverterResultExpression( + returnType, + filter.getRowType.getFieldNames) + s""" |${condition.code} |if (${condition.resultTerm}) { diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala index d747ba902bd27e..5b1c76375fe386 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala @@ -51,7 +51,10 @@ class DataSetProjectRule val generator = new CodeGenerator(config, inputType) // projection and implicit type conversion - val projection = generator.generateResultExpression(returnType, proj.getProjects) + val projection = generator.generateResultExpression( + returnType, + proj.getRowType.getFieldNames, + proj.getProjects) val body = s""" 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 492596c707657d..e1c9e96ec6c42b 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 @@ -18,12 +18,18 @@ package org.apache.flink.api.java.table.test; +import org.apache.flink.api.common.typeinfo.BasicTypeInfo; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.java.tuple.Tuple3; +import org.apache.flink.api.java.tuple.Tuple4; +import org.apache.flink.api.java.typeutils.TupleTypeInfo; import org.apache.flink.api.table.Table; import org.apache.flink.api.table.Row; import org.apache.flink.api.table.codegen.CodeGenException; import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.table.TableEnvironment; +import org.apache.flink.api.table.test.TableProgramsTestBase; import org.apache.flink.test.javaApiOperators.util.CollectionDataSets; import org.apache.flink.test.util.MultipleProgramsTestBase; import org.junit.Test; @@ -34,16 +40,16 @@ import java.util.List; @RunWith(Parameterized.class) -public class AsITCase extends MultipleProgramsTestBase { +public class AsITCase extends TableProgramsTestBase { - public AsITCase(TestExecutionMode mode){ - super(mode); + public AsITCase(TestExecutionMode mode, TableConfigMode configMode){ + super(mode, configMode); } @Test public void testAsFromTuple() throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); - TableEnvironment tableEnv = new TableEnvironment(); + TableEnvironment tableEnv = getJavaTableEnvironment(); Table table = tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c"); @@ -60,10 +66,54 @@ public void testAsFromTuple() throws Exception { compareResultAsText(results, expected); } - @Test(expected = CodeGenException.class) + @Test + public void testAsFromAndToTuple() throws Exception { + ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); + TableEnvironment tableEnv = getJavaTableEnvironment(); + + Table table = + tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c"); + + TypeInformation ti = new TupleTypeInfo>( + BasicTypeInfo.INT_TYPE_INFO, + BasicTypeInfo.LONG_TYPE_INFO, + BasicTypeInfo.STRING_TYPE_INFO); + + DataSet ds = tableEnv.toDataSet(table, ti); + List 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"; + compareResultAsText(results, expected); + } + + @Test + public void testAsFromTupleToPojo() throws Exception { + ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); + TableEnvironment tableEnv = getJavaTableEnvironment(); + + List> data = new ArrayList<>(); + data.add(new Tuple4<>("Rofl", 1, 1.0, "Hi")); + 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"); + + DataSet ds = tableEnv.toDataSet(table, SmallPojo2.class); + List results = ds.collect(); + String expected = "Rofl,1,1.0,Hi\n" + "lol,2,1.0,Hi\n" + "Test me,4,3.33,Hello world\n"; + compareResultAsText(results, expected); + } + + @Test public void testAsFromPojo() throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); - TableEnvironment tableEnv = new TableEnvironment(); + TableEnvironment tableEnv = getJavaTableEnvironment(); List data = new ArrayList<>(); data.add(new SmallPojo("Peter", 28, 4000.00, "Sales")); @@ -86,10 +136,88 @@ public void testAsFromPojo() throws Exception { compareResultAsText(results, expected); } + @Test + public void testAsFromPrivateFieldsPojo() throws Exception { + ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); + TableEnvironment tableEnv = getJavaTableEnvironment(); + + List data = new ArrayList<>(); + data.add(new PrivateSmallPojo("Peter", 28, 4000.00, "Sales")); + 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), + "department AS a, " + + "age AS b, " + + "salary AS c, " + + "name AS d"); + + DataSet ds = tableEnv.toDataSet(table, Row.class); + List 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); + } + + @Test + public void testAsFromAndToPojo() throws Exception { + ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); + TableEnvironment tableEnv = getJavaTableEnvironment(); + + List 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 ds = tableEnv.toDataSet(table, SmallPojo2.class); + List 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); + } + + @Test + public void testAsFromAndToPrivateFieldPojo() throws Exception { + ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); + TableEnvironment tableEnv = getJavaTableEnvironment(); + + List data = new ArrayList<>(); + data.add(new PrivateSmallPojo("Peter", 28, 4000.00, "Sales")); + 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), + "department AS a, " + + "age AS b, " + + "salary AS c, " + + "name AS d"); + + DataSet ds = tableEnv.toDataSet(table, PrivateSmallPojo2.class); + List 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); + } + @Test(expected = IllegalArgumentException.class) public void testAsWithToFewFields() throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); - TableEnvironment tableEnv = new TableEnvironment(); + TableEnvironment tableEnv = getJavaTableEnvironment(); Table table = tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b"); @@ -103,7 +231,7 @@ public void testAsWithToFewFields() throws Exception { @Test(expected = IllegalArgumentException.class) public void testAsWithToManyFields() throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); - TableEnvironment tableEnv = new TableEnvironment(); + TableEnvironment tableEnv = getJavaTableEnvironment(); Table table = tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c, d"); @@ -117,7 +245,7 @@ public void testAsWithToManyFields() throws Exception { @Test(expected = IllegalArgumentException.class) public void testAsWithAmbiguousFields() throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); - TableEnvironment tableEnv = new TableEnvironment(); + TableEnvironment tableEnv = getJavaTableEnvironment(); Table table = tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, b"); @@ -131,7 +259,7 @@ public void testAsWithAmbiguousFields() throws Exception { @Test(expected = IllegalArgumentException.class) public void testAsWithNonFieldReference1() throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); - TableEnvironment tableEnv = new TableEnvironment(); + TableEnvironment tableEnv = getJavaTableEnvironment(); Table table = tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a + 1, b, c"); @@ -145,7 +273,7 @@ public void testAsWithNonFieldReference1() throws Exception { @Test(expected = IllegalArgumentException.class) public void testAsWithNonFieldReference2() throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); - TableEnvironment tableEnv = new TableEnvironment(); + TableEnvironment tableEnv = getJavaTableEnvironment(); Table table = tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a as foo, b," + @@ -157,6 +285,8 @@ public void testAsWithNonFieldReference2() throws Exception { compareResultAsText(results, expected); } + // -------------------------------------------------------------------------------------------- + public static class SmallPojo { public SmallPojo() { } @@ -174,5 +304,130 @@ public SmallPojo(String name, int age, double salary, String department) { public String department; } + public static class PrivateSmallPojo { + + public PrivateSmallPojo() { } + + public PrivateSmallPojo(String name, int age, double salary, String department) { + this.name = name; + this.age = age; + this.salary = salary; + this.department = department; + } + + private String name; + private int age; + private double salary; + private String department; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public int getAge() { + return age; + } + + public void setAge(int age) { + this.age = age; + } + + public double getSalary() { + return salary; + } + + public void setSalary(double salary) { + this.salary = salary; + } + + public String getDepartment() { + return department; + } + + public void setDepartment(String department) { + this.department = department; + } + } + + public static class SmallPojo2 { + + public SmallPojo2() { } + + public SmallPojo2(String a, int b, double c, String d) { + this.a = a; + this.b = b; + this.c = c; + this.d = d; + } + + public String a; + public int b; + public double c; + public String d; + + @Override + public String toString() { + return a + "," + b + "," + c + "," + d; + } + } + + public static class PrivateSmallPojo2 { + + public PrivateSmallPojo2() { } + + public PrivateSmallPojo2(String a, int b, double c, String d) { + this.a = a; + this.b = b; + this.c = c; + this.d = d; + } + + private String a; + private int b; + private double c; + private String d; + + public String getA() { + return a; + } + + public void setA(String a) { + this.a = a; + } + + public int getB() { + return b; + } + + public void setB(int b) { + this.b = b; + } + + public double getC() { + return c; + } + + public void setC(double c) { + this.c = c; + } + + public String getD() { + return d; + } + + public void setD(String d) { + this.d = d; + } + + @Override + public String toString() { + return a + "," + b + "," + c + "," + d; + } + } + } 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 6779d4c030548b..54580c42563bfa 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 @@ -18,12 +18,14 @@ package org.apache.flink.api.scala.table.test -import org.apache.flink.api.table.Row import org.apache.flink.api.scala._ import org.apache.flink.api.scala.table._ import org.apache.flink.api.scala.util.CollectionDataSets -import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase} +import org.apache.flink.api.table.Row +import org.apache.flink.api.table.test.TableProgramsTestBase +import org.apache.flink.api.table.test.TableProgramsTestBase.TableConfigMode import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode +import org.apache.flink.test.util.TestBaseUtils import org.junit._ import org.junit.runner.RunWith import org.junit.runners.Parameterized @@ -31,7 +33,10 @@ import org.junit.runners.Parameterized import scala.collection.JavaConverters._ @RunWith(classOf[Parameterized]) -class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) { +class AsITCase( + mode: TestExecutionMode, + configMode: TableConfigMode) + extends TableProgramsTestBase(mode, configMode) { @Test def testAs(): Unit = { @@ -45,12 +50,11 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) { "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() + val results = t.toDataSet[Row](getConfig).collect() TestBaseUtils.compareResultAsText(results.asJava, expected) } @Test - @throws(classOf[Exception]) def testAsFromCaseClass(): Unit = { val env = ExecutionEnvironment.getExecutionEnvironment @@ -65,7 +69,26 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) { "Peter,28,4000.0,Sales\n" + "Anna,56,10000.0,Engineering\n" + "Lucy,42,6000.0,HR\n" - val results = t.toDataSet[Row].collect() + val results = t.toDataSet[Row](getConfig).collect() + TestBaseUtils.compareResultAsText(results.asJava, expected) + } + + @Test + def testAsFromAndToCaseClass(): 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 = + "SomeCaseClass(Peter,28,4000.0,Sales)\n" + + "SomeCaseClass(Anna,56,10000.0,Engineering)\n" + + "SomeCaseClass(Lucy,42,6000.0,HR)\n" + val results = t.toDataSet[SomeCaseClass](getConfig).collect() TestBaseUtils.compareResultAsText(results.asJava, expected) } @@ -76,7 +99,7 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) { val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b) val expected = "no" - val results = t.toDataSet[Row].collect() + val results = t.toDataSet[Row](getConfig).collect() TestBaseUtils.compareResultAsText(results.asJava, expected) } @@ -87,7 +110,7 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) { val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c, 'd) val expected = "no" - val results = t.toDataSet[Row].collect() + val results = t.toDataSet[Row](getConfig).collect() TestBaseUtils.compareResultAsText(results.asJava, expected) } @@ -98,7 +121,7 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) { val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'b) val expected = "no" - val results = t.toDataSet[Row].collect() + val results = t.toDataSet[Row](getConfig).collect() TestBaseUtils.compareResultAsText(results.asJava, expected) } @@ -110,7 +133,7 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) { val t = CollectionDataSets.get3TupleDataSet(env).as('a + 1, 'b, 'b) val expected = "no" - val results = t.toDataSet[Row].collect() + val results = t.toDataSet[Row](getConfig).collect() TestBaseUtils.compareResultAsText(results.asJava, expected) } @@ -122,7 +145,7 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) { val t = CollectionDataSets.get3TupleDataSet(env).as('a as 'foo, 'b, 'b) val expected = "no" - val results = t.toDataSet[Row].collect() + val results = t.toDataSet[Row](getConfig).collect() TestBaseUtils.compareResultAsText(results.asJava, expected) } From 8ffdbf922ff4ab32c0b5d7223a0bcd0eea59f37d Mon Sep 17 00:00:00 2001 From: twalthr Date: Sat, 13 Feb 2016 11:26:42 +0100 Subject: [PATCH 2/2] Feedback implemented --- .../api/java/table/TableEnvironment.scala | 14 +++++---- .../api/scala/table/TableEnvironment.scala | 7 +++-- .../api/table/codegen/CodeGenerator.scala | 29 +++++++++++++++---- 3 files changed, 36 insertions(+), 14 deletions(-) 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 edb3723e8b6d1b..938c778210d44d 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 @@ -65,9 +65,10 @@ class TableEnvironment { /** * 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 - * fields and the types must match. + * a DataSet. The given type must have exactly the same field types and field order as the + * [[org.apache.flink.api.table.Table]]. Row and tuple types can be mapped by position. + * POJO types require name equivalence to be mapped correctly as their fields do not have + * an order. */ @SuppressWarnings(Array("unchecked")) def toDataSet[T](table: Table, clazz: Class[T]): DataSet[T] = { @@ -77,9 +78,10 @@ class TableEnvironment { /** * 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 - * fields and the types must match. + * a DataSet. The given type must have exactly the same field types and field order as the + * [[org.apache.flink.api.table.Table]]. Row and tuple types can be mapped by position. + * POJO types require name equivalence to be mapped correctly as their fields do not have + * an order. */ def toDataSet[T](table: Table, typeInfo: TypeInformation[T]): DataSet[T] = { new JavaBatchTranslator(config).translate[T](table.relNode)(typeInfo) diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableEnvironment.scala index a05bb484346052..705378a843df4e 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableEnvironment.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableEnvironment.scala @@ -63,9 +63,10 @@ class TableEnvironment { /** * 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 - * fields and the types must match. + * a DataSet. The given type must have exactly the same field types and field order as the + * [[org.apache.flink.api.table.Table]]. Row and tuple types can be mapped by position. + * POJO types require name equivalence to be mapped correctly as their fields do not have + * an order. */ def toDataSet[T: TypeInformation](table: Table): DataSet[T] = { new ScalaBatchTranslator(config).translate[T](table.relNode) diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala index 1f71b622618576..b6ef49b4bb1f7f 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala @@ -50,9 +50,24 @@ class CodeGenerator( config: TableConfig, input1: TypeInformation[Any], input2: Option[TypeInformation[Any]] = None, - inputPojoFieldMapping: Array[Int] = Array()) + inputPojoFieldMapping: Option[Array[Int]] = None) extends RexVisitor[GeneratedExpression] { + // check for POJO input mapping + input1 match { + case pt: PojoTypeInfo[_] => + inputPojoFieldMapping.getOrElse( + throw new CodeGenException("No input mapping is specified for input of type POJO.")) + case _ => // ok + } + + // check that input2 is never a POJO + input2 match { + case pt: PojoTypeInfo[_] => + throw new CodeGenException("Second input must not be a POJO type.") + case _ => // ok + } + /** * A code generator for generating unary Flink * [[org.apache.flink.api.common.functions.Function]]s with one input. @@ -63,7 +78,7 @@ class CodeGenerator( * POJO (POJO types have no deterministic field order). */ def this(config: TableConfig, input: TypeInformation[Any], inputPojoFieldMapping: Array[Int]) = - this(config, input, None, inputPojoFieldMapping) + this(config, input, None, Some(inputPojoFieldMapping)) // set of member statements that will be added only once @@ -266,6 +281,10 @@ class CodeGenerator( if (returnType.getArity != fieldExprs.length) { throw new CodeGenException("Arity of result type does not match number of expressions.") } + if (resultFieldNames.length != fieldExprs.length) { + throw new CodeGenException("Arity of result field names does not match number of " + + "expressions.") + } // type check returnType match { case pt: PojoTypeInfo[_] => @@ -703,7 +722,7 @@ class CodeGenerator( case ct: CompositeType[_] => val fieldIndex = if (ct.isInstanceOf[PojoTypeInfo[_]]) { - inputPojoFieldMapping(index) + inputPojoFieldMapping.get(index) } else { index @@ -855,7 +874,7 @@ class CodeGenerator( val statement = ti match { case rt: RowTypeInfo => s""" - |${ti.getTypeClass.getCanonicalName} $outRecordTerm = + |transient ${ti.getTypeClass.getCanonicalName} $outRecordTerm = | new ${ti.getTypeClass.getCanonicalName}(${rt.getArity}); |""".stripMargin case _ => @@ -871,7 +890,7 @@ class CodeGenerator( val fieldTerm = s"field_${clazz.getCanonicalName.replace('.', '$')}_$fieldName" val fieldExtraction = s""" - |java.lang.reflect.Field $fieldTerm = + |transient java.lang.reflect.Field $fieldTerm = | org.apache.flink.api.java.typeutils.TypeExtractor.getDeclaredField( | ${clazz.getCanonicalName}.class, "$fieldName"); |""".stripMargin