[FLINK-12200] [Table-planner] Support UNNEST for MAP types #8179

art4ul · 2019-04-15T19:50:08Z

What is the purpose of the change

This pull request adds support of UNNEST operator for MAP types.

Brief change log

The LogicalUnnestRule now handle the MapRelDataType and create appropriate ExplodeTableFunction for it

Verifying this change

This change added tests and can be verified as follows:

Added test that validates that the following SQL query works correctly for the table with MAP type
SELECT a,b,v FROM src CROSS JOIN UNNEST(c) as f (k,v)

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? not applicable

Syncing the fork to the original repository

Sync

flinkbot · 2019-04-15T19:51:26Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

Details

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

art4ul · 2019-04-18T09:55:25Z

Guys! Could you please review this pull request?

art4ul · 2019-04-22T12:12:38Z

@KurtYoung @JingsongLi Could you please review this pull request or connect to guys who responsible for this component? Thanks in advance!

KurtYoung

Thanks for working on this, I left some comments. BTW, i noticed there are some format issues and i'm not sure i pointed out all of them, please check it.

KurtYoung · 2019-04-22T13:03:59Z

...ble-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala

+            case map: MapRelDataType =>
+              val keyTypeInfo = FlinkTypeFactory.toTypeInfo(map.keyType)
+              val valueTypeInfo = FlinkTypeFactory.toTypeInfo(map.valueType)
+              val componentTypeInfo = createTuple2TypeInformation(keyTypeInfo,valueTypeInfo)


space after comma

Is it more appropriate for us to return RowType here?

space after comma

Fixed

Is it more appropriate for us to return RowType here?

I need to return typeInfo for an exploded type which will be as a field in the resultant Row. I'm going to use the following transformation form Map[K,V] into a sequence of tuples [(K,V), ...]. I need it to reuse the SQL syntaxis of mapping result on the tuple. I mean I want to support the following SQL query:

SELECT k, v FROM t1 , UNNEST(t1.c) as A (k,v)

This is the reason why I return TupleTypeInfo

I checked other system's behavior, it seems more common to unnest map field into two columns, in your case, it would be:

SELECT k, v FROM t1 , UNNEST(t1.c) as (k,v)

@KurtYoung I'm sorry but in case if I use SQL like in your example I got org.apache.flink.table.api.SqlParserException in method org.apache.flink.table.calcite.FlinkPlannerImpl.parse

Sorry for giving a wrong example. What i want to say is we should make built-in table functions looks more easy and explicit for framework. To this very example, you relied on some implicit functionality which Flink currently support. The first one is you replied on Flink can implicitly convert Tuple to a Row, and the second one is you don't provide any valuable type related information in your table function, all the type information only provided in the logical rule.

I'm not saying this is wrong, but i think there is another way to make all these things more accurate and explicit. For example, you can explicitly tell framework your table function is returning Row type, and give more type related information to framework. It will make these codes more robust, have less possibility break in the future. Can you give this a try? If this will involve lots of changes, i'm also ok with current version.

@KurtYoung I got it, thank you a lot for your explanation. I return the ROW type now instead of Tuple as was previously.

KurtYoung · 2019-04-22T13:04:47Z

...ble-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala

+              val valueTypeInfo = FlinkTypeFactory.toTypeInfo(map.valueType)
+              val componentTypeInfo = createTuple2TypeInformation(keyTypeInfo,valueTypeInfo)
+              val componentType = cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
+                .createTypeFromTypeInfo(componentTypeInfo,true)


space after comma

KurtYoung · 2019-04-22T13:05:08Z

...ble-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala

+                .createTypeFromTypeInfo(componentTypeInfo,true)
+
+              val explodeFunction = ExplodeFunctionUtil.explodeTableFuncFromType(map.typeInfo)
+              (componentType , explodeFunction)


extra space before comma

KurtYoung · 2019-04-22T13:05:31Z

...link-table-planner/src/main/scala/org/apache/flink/table/plan/util/ExplodeFunctionUtil.scala

  }
 }

+class MapExplodeTableFunc extends TableFunction[Object]{


space before brace

I'm not sure making this table function returning object and let framework rely on information somewhere else to perform right is good choice. This will make the codes fragile.

space before brace

fixed

I'm not sure making this table function returning object and let framework rely on information somewhere else to perform right is good choice. This will make the codes fragile.

The UNNEST function should be able to explode any type of key and value types. This is the reason why I'm using Object type here. Basically, I use the same approach which used for Array types (for the arrays used the ObjectExplodeTableFunc)

ObjectExplodeTableFunc is not directly used by all arrays which want to be unnested. It's only used by the array which has object array element type, .e.g. Array(object[]).

Thanks for your comments, I've fixed the TableFunction and now it returns Row

JingsongLi

Thanks for working on this, basically LGTM. just some format issues.

JingsongLi · 2019-04-23T01:49:08Z

...link-table-planner/src/main/scala/org/apache/flink/table/plan/util/ExplodeFunctionUtil.scala

+class MapExplodeTableFunc extends TableFunction[Object] {
+
+  def eval(map: util.Map[Object, Object]): Unit = {
+    map.asScala.foreach{ case (key,value) =>


map.asScala.foreach { case (key, value)
space need here

JingsongLi · 2019-04-23T01:49:18Z

...link-table-planner/src/main/scala/org/apache/flink/table/plan/util/ExplodeFunctionUtil.scala

+
+  def eval(map: util.Map[Object, Object]): Unit = {
+    map.asScala.foreach{ case (key,value) =>
+      collect((key,value))


collect((key, value))

JingsongLi · 2019-04-23T02:02:57Z

The PR component tag should be [table-planner] instead of [Table API?

art4ul · 2019-04-23T10:04:12Z

The PR component tag should be [table-planner] instead of [Table API?

Thank you for your note, I've fixed the tag in the title .

art4ul · 2019-04-23T10:07:33Z

@KurtYoung @JingsongLi Thank you guys for your feedback. I've fixed all format issues and add some answers to your questions.

JingsongLi · 2019-04-23T10:24:09Z

LGTM +1

This closes apache#8179

art4ul added 3 commits January 10, 2019 18:51

Merge pull request #1 from apache/master

bbd2218

Syncing the fork to the original repository

Merge pull request #2 from apache/master

c32cdc8

Sync

FLINK-12200 [Table API] Support UNNEST for MAP types

a7b022b

rmetzger added the review=description? label Apr 15, 2019

Fix test

ce4e51a

rmetzger added the component=TableSQL/Planner label Apr 15, 2019

KurtYoung reviewed Apr 22, 2019

View reviewed changes

[FLINK-12252] Fix code style

67d9703

JingsongLi reviewed Apr 23, 2019

View reviewed changes

[FLINK-12252] Fix code style

acae1ed

art4ul changed the title ~~[FLINK-12200] [Table API] Support UNNEST for MAP types~~ [FLINK-12200] [Table-planner] Support UNNEST for MAP types Apr 23, 2019

art4ul added 2 commits April 26, 2019 12:19

[FLINK-12200] Use Row as return type in Map explode function

dfa664f

[FLINK-12200] Fix return type of Explode Function

7ccdddd

KurtYoung merged commit 2c39024 into apache:master Apr 26, 2019

tianchen92 pushed a commit to tianchen92/flink that referenced this pull request May 13, 2019

[FLINK-12200][table-planner] Support UNNEST for MAP types

1bf5f03

This closes apache#8179

[FLINK-12200] [Table-planner] Support UNNEST for MAP types #8179

[FLINK-12200] [Table-planner] Support UNNEST for MAP types #8179

Uh oh!

Conversation

art4ul commented Apr 15, 2019

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Apr 15, 2019

Review Progress

Uh oh!

art4ul commented Apr 18, 2019

Uh oh!

art4ul commented Apr 22, 2019

Uh oh!

KurtYoung left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

art4ul Apr 23, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JingsongLi left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JingsongLi commented Apr 23, 2019

Uh oh!

art4ul commented Apr 23, 2019

Uh oh!

art4ul commented Apr 23, 2019

Uh oh!

JingsongLi commented Apr 23, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

art4ul Apr 23, 2019 •

edited

Loading