[FLINK-11975][table-planner-blink] Support to run a sql query : 'SELECT * FROM (VALUES (1, 2, 3)) T(a, b, c)' #8035

beyond1920 · 2019-03-22T11:47:18Z

What is the purpose of the change

Support to run a sql query : 'SELECT * FROM (VALUES (1, 2, 3)) T(a, b, c)'

Brief change log

add writeToSink, translateNodeDag in batch and stream tableEnv
introduce SinkRules for batch and stream
Introduce subclass of TableSink, including BaseUpsertStreamTableSink, BatchTableSink, CollectTableSink, DataStreamTableSink
StreamExecSink/BatchExecSink implements ExecNode interface
StramExecValues/BatchExecValues implements ExecNode interface, add CodeGen for Values.
add Itcase test infrastructure, add Itcase to run SELECT * FROM (VALUES (1, 2, 3)) T(a, b, c)' for batch and stream

Verifying this change

ITCase

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? (yes)
If yes, how is the feature documented? (JavaDocs)

flinkbot · 2019-03-22T11:48:20Z

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.

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

JingsongLi

Thanks @beyond1920 for your pull request, I left some comments.

...flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/GeneratedClass.java

...able-runtime-blink/src/main/java/org/apache/flink/table/typeutils/TimeIndicatorTypeInfo.java

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/util/TypeUtils.java

...able/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/TypeConverters.java

...flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/GeneratedInput.java

...nk-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/util/BaseRowUtil.java

...le/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/GenericRow.java

...le-planner-blink/src/main/scala/org/apache/flink/table/sinks/BaseUpsertStreamTableSink.scala

JingsongLi · 2019-03-25T02:59:11Z

...le-planner-blink/src/main/scala/org/apache/flink/table/sinks/BaseUpsertStreamTableSink.scala

+  *
+  * @tparam T Type of records that this [[TableSink]] expects and supports.
+  */
+trait BaseUpsertStreamTableSink[T] extends StreamTableSink[T] {


Modify it to java Interface?

JingsongLi · 2019-03-25T02:59:42Z

...e/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/BatchTableSink.scala

+  *
+  * @tparam T Type of [[DataStream]] that this [[TableSink]] expects and supports.
+  */
+trait BatchTableSink[T] extends TableSink[T] {


Java interface?

...le-planner-blink/src/main/scala/org/apache/flink/table/sinks/BaseUpsertStreamTableSink.scala

JingsongLi · 2019-03-25T03:06:03Z

...-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/AppendStreamTableSink.scala

+  *
+  * @tparam T Type of [[DataStream]] that this [[TableSink]] expects and supports.
+  */
+trait AppendStreamTableSink[T] extends StreamTableSink[T] {


java interface?

...e-planner-blink/src/main/scala/org/apache/flink/table/codegen/InputFormatCodeGenerator.scala

...table-planner-blink/src/main/scala/org/apache/flink/table/codegen/CodeGeneratorContext.scala

...ink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala

flink-core/src/main/java/org/apache/flink/util/TimeConvertUtils.java

KurtYoung

Thanks for opening this PR! I think the codes need some more refinements, left some comments

KurtYoung · 2019-03-25T03:23:59Z

flink-core/src/main/java/org/apache/flink/util/TimeConvertUtils.java

+/**
+ * A {@link TimeConvertUtils} is used to convert Time Type.
+ */
+public class TimeConvertUtils {


I think we could keep this in table-planner-blink. And a further question is, do we really need this? What's the differences between this utility and DateTimeUtils?

TimeConvertUtils is to fix TimeZone problems, which is refactor in FLINK-11898, So temporally to throw an Exception here.

flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/util/StateUtil.java

...nk-table-planner-blink/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala

...nk-table-runtime-blink/src/main/java/org/apache/flink/table/generated/ValuesInputFormat.java

KurtYoung · 2019-03-25T03:44:43Z

...er-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSink.scala

+  /**
+    * Returns [[DamBehavior]] of this node.
+    */
+  override def getDamBehavior: DamBehavior = DamBehavior.PIPELINED


?? I don't catch your meaning.

The definition of PIPELINED in dam behavior is that the operator will not cache any data and all records will pass through in a pipelined fashion. For sink operator, the records will not pass through it, right? To me, the behavior is more like to be FULL_DAM, except the sink operator will not return or output any record after it consumes all the input data.

DamBehavior of sink is never used actually, so it's value will not effect schedule or deadlock breaking. However, Full_DAM is a more appropriate choice in this point of view, I will update it.

...er-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSink.scala

beyond1920 · 2019-03-25T09:45:25Z

@JingsongLi thanks for your comments. All done, except one minor question, TableSink interfaces still in Scala instead of Java, could do it later with all sinks and sources.

beyond1920 · 2019-03-25T09:47:31Z

@KurtYoung Thanks for your comments. All done, except two minor question:

BatchCompatibleStreamTableSink could be introduced later when we really need it?
override def getDamBehavior: DamBehavior = DamBehavior.PIPELINED I don't understand your comment here.

…CT * FROM (VALUES (1, 2, 3)) T(a, b, c)'

JingsongLi · 2019-03-26T06:34:23Z

+1 LGTM

2. Rename BaseUpsertStreamTableSink, BaseRetractStreamTableSink 3. Update codeStyle 4. other minor update based on comments

...link-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/ValuesInputFormat.java

...-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecSink.scala

2. Move valuesInputFormat to values subpackage 3. Minor update in StreamExecSink and BatchExecSink

…om values query (apache#8035)

rmetzger added review=description? component=TableSQL/Planner labels Mar 22, 2019

beyond1920 force-pushed the flink-11975 branch 5 times, most recently from dec0598 to 2386edf Compare March 25, 2019 02:37

JingsongLi requested changes Mar 25, 2019

View reviewed changes

JingsongLi reviewed Mar 25, 2019

View reviewed changes

KurtYoung requested changes Mar 25, 2019

View reviewed changes

JingsongLi reviewed Mar 25, 2019

View reviewed changes

...er-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSink.scala Show resolved Hide resolved

beyond1920 force-pushed the flink-11975 branch from 2386edf to 047471a Compare March 25, 2019 09:44

beyond1920 force-pushed the flink-11975 branch 3 times, most recently from 0b90cfc to b2f2f20 Compare March 25, 2019 11:06

[FLINK-11975][table-planner-blink] Support to run a sql query : 'SELE…

ab79ce2

…CT * FROM (VALUES (1, 2, 3)) T(a, b, c)'

beyond1920 force-pushed the flink-11975 branch 5 times, most recently from cb3f443 to 32a1fb3 Compare March 26, 2019 05:32

1. Remove StateUtil,TimeIndicatorTypeInfo, TypeUtils, TimeConvertUtils

2307305

2. Rename BaseUpsertStreamTableSink, BaseRetractStreamTableSink 3. Update codeStyle 4. other minor update based on comments

KurtYoung reviewed Mar 26, 2019

View reviewed changes

...link-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/ValuesInputFormat.java Outdated Show resolved Hide resolved

...-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecSink.scala Outdated Show resolved Hide resolved

beyond1920 force-pushed the flink-11975 branch 2 times, most recently from ea18552 to c6ce3fe Compare March 26, 2019 09:09

1. Revert UpsertStreamTableSink, RetractStreamTableSink

40b55c6

2. Move valuesInputFormat to values subpackage 3. Minor update in StreamExecSink and BatchExecSink

beyond1920 force-pushed the flink-11975 branch from c6ce3fe to 40b55c6 Compare March 26, 2019 09:15

KurtYoung approved these changes Mar 26, 2019

View reviewed changes

KurtYoung merged commit 6333fc7 into apache:master Mar 26, 2019

HuangZhenQiu pushed a commit to HuangZhenQiu/flink that referenced this pull request Apr 22, 2019

[FLINK-11975][table-planner-blink] Support running a simple select fr…

1e72f65

…om values query (apache#8035)

sunhaibotb pushed a commit to sunhaibotb/flink that referenced this pull request May 8, 2019

[FLINK-11975][table-planner-blink] Support running a simple select fr…

4906e83

…om values query (apache#8035)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-11975][table-planner-blink] Support to run a sql query : 'SELECT * FROM (VALUES (1, 2, 3)) T(a, b, c)' #8035

[FLINK-11975][table-planner-blink] Support to run a sql query : 'SELECT * FROM (VALUES (1, 2, 3)) T(a, b, c)' #8035

beyond1920 commented Mar 22, 2019

flinkbot commented Mar 22, 2019

JingsongLi left a comment

JingsongLi Mar 25, 2019

JingsongLi Mar 25, 2019

JingsongLi Mar 25, 2019

KurtYoung left a comment

KurtYoung Mar 25, 2019

beyond1920 Mar 25, 2019

KurtYoung Mar 25, 2019

beyond1920 Mar 25, 2019

KurtYoung Mar 26, 2019

beyond1920 Mar 26, 2019

beyond1920 commented Mar 25, 2019

beyond1920 commented Mar 25, 2019

JingsongLi commented Mar 26, 2019

[FLINK-11975][table-planner-blink] Support to run a sql query : 'SELECT * FROM (VALUES (1, 2, 3)) T(a, b, c)' #8035

[FLINK-11975][table-planner-blink] Support to run a sql query : 'SELECT * FROM (VALUES (1, 2, 3)) T(a, b, c)' #8035

Conversation

beyond1920 commented Mar 22, 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

flinkbot commented Mar 22, 2019

Review Progress

JingsongLi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KurtYoung left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

beyond1920 commented Mar 25, 2019

beyond1920 commented Mar 25, 2019

JingsongLi commented Mar 26, 2019