You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2018/02/06 07:30:42 UTC
flink git commit: [FLINK-7923] [table] Support accessing fields of
arrays of composite types
Repository: flink
Updated Branches:
refs/heads/master 3d323ba10 -> 91b7d0110
[FLINK-7923] [table] Support accessing fields of arrays of composite types
This closes #5367.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/91b7d011
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/91b7d011
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/91b7d011
Branch: refs/heads/master
Commit: 91b7d011022fc1211aebe9114dfa6b497da49553
Parents: 3d323ba
Author: Shuyi Chen <sh...@uber.com>
Authored: Thu Jan 25 15:36:36 2018 -0800
Committer: twalthr <tw...@apache.org>
Committed: Tue Feb 6 08:18:17 2018 +0100
----------------------------------------------------------------------
.../flink/table/codegen/CodeGenerator.scala | 86 ++++++++++++--------
.../table/expressions/CompositeAccessTest.scala | 43 ++++++++++
.../utils/CompositeTypeTestBase.scala | 18 +++-
3 files changed, 111 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/91b7d011/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
index aca0ba3..756a828 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
@@ -581,39 +581,7 @@ abstract class CodeGenerator(
override def visitFieldAccess(rexFieldAccess: RexFieldAccess): GeneratedExpression = {
val refExpr = rexFieldAccess.getReferenceExpr.accept(this)
val index = rexFieldAccess.getField.getIndex
- val fieldAccessExpr = generateFieldAccess(
- refExpr.resultType,
- refExpr.resultTerm,
- index)
-
- val resultTerm = newName("result")
- val nullTerm = newName("isNull")
- val resultTypeTerm = primitiveTypeTermForTypeInfo(fieldAccessExpr.resultType)
- val defaultValue = primitiveDefaultValue(fieldAccessExpr.resultType)
- val resultCode = if (nullCheck) {
- s"""
- |${refExpr.code}
- |$resultTypeTerm $resultTerm;
- |boolean $nullTerm;
- |if (${refExpr.nullTerm}) {
- | $resultTerm = $defaultValue;
- | $nullTerm = true;
- |}
- |else {
- | ${fieldAccessExpr.code}
- | $resultTerm = ${fieldAccessExpr.resultTerm};
- | $nullTerm = ${fieldAccessExpr.nullTerm};
- |}
- |""".stripMargin
- } else {
- s"""
- |${refExpr.code}
- |${fieldAccessExpr.code}
- |$resultTypeTerm $resultTerm = ${fieldAccessExpr.resultTerm};
- |""".stripMargin
- }
-
- GeneratedExpression(resultTerm, nullTerm, resultCode, fieldAccessExpr.resultType)
+ generateFieldAccess(refExpr, index)
}
override def visitLiteral(literal: RexLiteral): GeneratedExpression = {
@@ -1001,6 +969,21 @@ abstract class CodeGenerator(
requireArray(array)
generateArrayElement(this, array)
+ case DOT =>
+ // Due to https://issues.apache.org/jira/browse/CALCITE-2162, expression such as
+ // "array[1].a.b" won't work now.
+ if (operands.size > 2) {
+ throw new CodeGenException(
+ "A DOT operator with more than 2 operands is not supported yet.")
+ }
+ val fieldName = call.operands.get(1).asInstanceOf[RexLiteral].getValueAs(classOf[String])
+ val fieldIdx = operands
+ .head
+ .resultType
+ .asInstanceOf[CompositeType[_]]
+ .getFieldIndex(fieldName)
+ generateFieldAccess(operands.head, fieldIdx)
+
case ScalarSqlFunctions.CONCAT =>
generateConcat(this.nullCheck, operands)
@@ -1041,6 +1024,43 @@ abstract class CodeGenerator(
// generator helping methods
// ----------------------------------------------------------------------------------------------
+ private def generateFieldAccess(refExpr: GeneratedExpression, index: Int): GeneratedExpression = {
+
+ val fieldAccessExpr = generateFieldAccess(
+ refExpr.resultType,
+ refExpr.resultTerm,
+ index)
+
+ val resultTerm = newName("result")
+ val nullTerm = newName("isNull")
+ val resultTypeTerm = primitiveTypeTermForTypeInfo(fieldAccessExpr.resultType)
+ val defaultValue = primitiveDefaultValue(fieldAccessExpr.resultType)
+ val resultCode = if (nullCheck) {
+ s"""
+ |${refExpr.code}
+ |$resultTypeTerm $resultTerm;
+ |boolean $nullTerm;
+ |if (${refExpr.nullTerm}) {
+ | $resultTerm = $defaultValue;
+ | $nullTerm = true;
+ |}
+ |else {
+ | ${fieldAccessExpr.code}
+ | $resultTerm = ${fieldAccessExpr.resultTerm};
+ | $nullTerm = ${fieldAccessExpr.nullTerm};
+ |}
+ |""".stripMargin
+ } else {
+ s"""
+ |${refExpr.code}
+ |${fieldAccessExpr.code}
+ |$resultTypeTerm $resultTerm = ${fieldAccessExpr.resultTerm};
+ |""".stripMargin
+ }
+
+ GeneratedExpression(resultTerm, nullTerm, resultCode, fieldAccessExpr.resultType)
+ }
+
private def generateInputAccess(
inputType: TypeInformation[_ <: Any],
inputTerm: String,
http://git-wip-us.apache.org/repos/asf/flink/blob/91b7d011/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/CompositeAccessTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/CompositeAccessTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/CompositeAccessTest.scala
index bd1c199..ca9a7ab 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/CompositeAccessTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/CompositeAccessTest.scala
@@ -119,6 +119,49 @@ class CompositeAccessTest extends CompositeTypeTestBase {
testTableApi(12.flatten(), "12.flatten()", "12")
testTableApi('f5.flatten(), "f5.flatten()", "13")
+
+ // array of composites
+ testAllApis(
+ 'f8.at(1).get("_1"),
+ "f8.at(1).get('_1')",
+ "f8[1]._1",
+ "true"
+ )
+
+ testAllApis(
+ 'f8.at(1).get("_2"),
+ "f8.at(1).get('_2')",
+ "f8[1]._2",
+ "23"
+ )
+
+ testAllApis(
+ 'f9.at(2).get("_1"),
+ "f9.at(2).get('_1')",
+ "f9[2]._1",
+ "null"
+ )
+
+ testAllApis(
+ 'f10.at(1).get("stringField"),
+ "f10.at(1).get('stringField')",
+ "f10[1].stringField",
+ "Bob"
+ )
+
+ testAllApis(
+ 'f11.at(1).get("myString"),
+ "f11.at(1).get('myString')",
+ "f11[1].myString",
+ "Hello"
+ )
+
+ testAllApis(
+ 'f12.at(1).get("arrayField").at(1).get("stringField"),
+ "f12.at(1).get('arrayField').at(1).get('stringField')",
+ "f12[1].arrayField[1].stringField",
+ "Alice"
+ )
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/91b7d011/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/CompositeTypeTestBase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/CompositeTypeTestBase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/CompositeTypeTestBase.scala
index 8f7360e..c214109 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/CompositeTypeTestBase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/CompositeTypeTestBase.scala
@@ -22,13 +22,13 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils.{RowTypeInfo, TupleTypeInfo, TypeExtractor}
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.table.api.Types
-import org.apache.flink.table.expressions.utils.CompositeTypeTestBase.{MyCaseClass, MyCaseClass2, MyPojo}
+import org.apache.flink.table.expressions.utils.CompositeTypeTestBase.{MyCaseClass, MyCaseClass2, MyCaseClass3, MyPojo}
import org.apache.flink.types.Row
class CompositeTypeTestBase extends ExpressionTestBase {
def testData: Row = {
- val testData = new Row(8)
+ val testData = new Row(13)
testData.setField(0, MyCaseClass(42, "Bob", booleanField = true))
testData.setField(1, MyCaseClass2(MyCaseClass(25, "Timo", booleanField = false)))
testData.setField(2, ("a", "b"))
@@ -37,6 +37,11 @@ class CompositeTypeTestBase extends ExpressionTestBase {
testData.setField(5, 13)
testData.setField(6, MyCaseClass2(null))
testData.setField(7, Tuple1(true))
+ testData.setField(8, Array(Tuple2(true, 23), Tuple2(false, 12)))
+ testData.setField(9, Array(Tuple1(true), null))
+ testData.setField(10, Array(MyCaseClass(42, "Bob", booleanField = true)))
+ testData.setField(11, Array(new MyPojo()))
+ testData.setField(12, Array(MyCaseClass3(Array(MyCaseClass(42, "Alice", booleanField = true)))))
testData
}
@@ -49,7 +54,12 @@ class CompositeTypeTestBase extends ExpressionTestBase {
TypeExtractor.createTypeInfo(classOf[MyPojo]),
Types.INT,
createTypeInformation[MyCaseClass2],
- createTypeInformation[Tuple1[Boolean]]
+ createTypeInformation[Tuple1[Boolean]],
+ createTypeInformation[Array[Tuple2[Boolean, Int]]],
+ createTypeInformation[Array[Tuple1[Boolean]]],
+ createTypeInformation[Array[MyCaseClass]],
+ createTypeInformation[Array[MyPojo]],
+ createTypeInformation[Array[MyCaseClass3]]
).asInstanceOf[TypeInformation[Any]]
}
}
@@ -59,6 +69,8 @@ object CompositeTypeTestBase {
case class MyCaseClass2(objectField: MyCaseClass)
+ case class MyCaseClass3(arrayField: Array[MyCaseClass])
+
class MyPojo {
private var myInt: Int = 0
private var myString: String = "Hello"