[FLINK-18726][table-planner-blink] Support INSERT INTO specific colum… #14977

docete · 2021-02-22T03:21:15Z

…ns in blink planner

What is the purpose of the change

This PR supports INSERT INTO specific columns in blink planner.

Brief change log

aa8dede: Support INSERT INTO specific columns in blink planner

Verifying this change

This change added tests.

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

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

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2021-02-22T03:24:07Z

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.

Automated Checks

Last check on commit aa8dede (Mon Feb 22 03:24:06 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

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

flinkbot · 2021-02-22T03:41:23Z

CI report:

f192b80 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

wuchong · 2021-02-22T04:08:22Z

cc @leonardBang

leonardBang

Thanks @docete for the contribution, the PR generally LGTM, I only left minor comment about exception message, could you help update it?

leonardBang · 2021-02-22T10:05:25Z

...ink/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/TableSinkITCase.scala

+ tEnv.createTemporaryView("MyTable", t)
+
+ expectedEx.expect(classOf[ValidationException])
+ expectedEx.expectMessage("Column 'a' has no default value and does not allow NULLs")


Flink does not support default value yet, we can improve exception message here, how about improve to

Suggested change

expectedEx.expectMessage("Column 'a' has no default value and does not allow NULLs")

expectedEx.expectMessage("Column 'a' does not allow NULL value, thus it can not be used in partial insert")

sounds good, will update soon.

Check the error msg again and found the original one is more accuracy. So i will revert the change.

sllence · 2021-02-24T04:09:29Z

https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/sqlexec/SqlToOperationConverter.java

else if (validated instanceof RichSqlInsert) {
            SqlNodeList targetColumnList = ((RichSqlInsert) validated).getTargetColumnList();
            if (targetColumnList != null && targetColumnList.size() != 0) {
                throw new ValidationException("Partial inserts are not supported");
            }
            return Optional.of(converter.convertSqlInsert((RichSqlInsert) validated));
        }

Does it need to be changed here？

docete

revert the change and rebase master

docete · 2021-02-24T06:38:23Z

...ink/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/TableSinkITCase.scala

+ tEnv.createTemporaryView("MyTable", t)
+
+ expectedEx.expect(classOf[ValidationException])
+ expectedEx.expectMessage("Column 'a' has no default value and does not allow NULLs")


Check the error msg again and found the original one is more accuracy. So i will revert the change.

leonardBang

Thanks @docete for the update, LGTM

leonardBang · 2021-02-24T07:34:54Z

https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/sqlexec/SqlToOperationConverter.java
else if (validated instanceof RichSqlInsert) {
            SqlNodeList targetColumnList = ((RichSqlInsert) validated).getTargetColumnList();
            if (targetColumnList != null && targetColumnList.size() != 0) {
                throw new ValidationException("Partial inserts are not supported");
            }
            return Optional.of(converter.convertSqlInsert((RichSqlInsert) validated));
        }
Does it need to be changed here？

Nice catch, I think we can remove this check safely . @docete

sllence · 2021-02-24T09:25:03Z

https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/sqlexec/SqlToOperationConverter.java
else if (validated instanceof RichSqlInsert) {
            SqlNodeList targetColumnList = ((RichSqlInsert) validated).getTargetColumnList();
            if (targetColumnList != null && targetColumnList.size() != 0) {
                throw new ValidationException("Partial inserts are not supported");
            }
            return Optional.of(converter.convertSqlInsert((RichSqlInsert) validated));
        }
Does it need to be changed here？
Nice catch, I think we can remove this check safely . @docete

Sorry，I've looked at it again. This verification is in the old planner. It should not be modified

docete · 2021-02-25T06:53:42Z

@flinkbot run azure

wuchong · 2021-02-25T08:40:42Z

...link/src/test/scala/org/apache/flink/table/planner/runtime/batch/table/TableSinkITCase.scala

+
+ tEnv.executeSql(
+ s"""
+ |INSERT INTO testSink (c, d, e)


Could you add a test which reorder the specified columns, e.g. INSERT INTO .. (e, c, d). I tried in my local machine and it fails.

…ns in blink planner

wuchong

LGTM.

wuchong · 2021-03-01T11:40:33Z

The failed e2e is because of time out.
Will merge this PR.

JingsongLi · 2021-03-10T08:14:11Z

...lanner-blink/src/main/scala/org/apache/flink/table/planner/calcite/PreValidateReWriter.scala

+ targetPosition: util.List[Int]): util.ArrayList[SqlNode] = {
+ val targetList = new Array[SqlNode](sourceList.size())
+ 0 until sourceList.size() foreach {
+ idx => targetList(targetPosition.get(idx)) = sourceList.get(idx)


Here should be idx => targetList(idx) = sourceList.get(targetPosition.get(idx)) What do you think? @leonardBang @wuchong

Is there any tests to prove this?

@Test def testPartialInsertWithComplexReorder(): Unit = { tEnv.executeSql( s""" |CREATE TABLE testSink ( | `a` INT, | `c` STRING, | `c1` STRING, | `c2` STRING, | `c3` BIGINT, | `d` INT, | `e` DOUBLE |) |WITH ( | 'connector' = 'values', | 'sink-insert-only' = 'false' |) |""".stripMargin) val t = env.fromCollection(tupleData2).toTable(tEnv, 'x, 'y) tEnv.createTemporaryView("MyTable", t) tEnv.executeSql( s""" |INSERT INTO testSink (a,c2,e,c,c1,c3,d) |SELECT 1,'c2',sum(y),'c','c1',33333,12 FROM MyTable GROUP BY x |""".stripMargin).await() val expected = List( "1,c,c1,c2,33333,12,0.1", "1,c,c1,c2,33333,12,0.4", "1,c,c1,c2,33333,12,1.0", "1,c,c1,c2,33333,12,2.2", "1,c,c1,c2,33333,12,3.9") val result = TestValuesTableFactory.getResults("testSink") assertEquals(expected.sorted, result.sorted) }

BTW, I think we should use plan testing instead of itcase.

rmetzger added review=description? component=TableSQL/Planner labels Feb 22, 2021

leonardBang reviewed Feb 22, 2021

View reviewed changes

docete force-pushed the FLINK-18726 branch from 9b596a1 to cbb43f9 Compare February 24, 2021 06:32

docete commented Feb 24, 2021

View reviewed changes

leonardBang approved these changes Feb 24, 2021

View reviewed changes

wuchong reviewed Feb 25, 2021

View reviewed changes

docete added 2 commits March 1, 2021 13:11

[FLINK-18726][table-planner-blink] Support INSERT INTO specific colum…

555a06f

…ns in blink planner

fixup: fix reorder issues of partial insert

f192b80

docete force-pushed the FLINK-18726 branch from cbb43f9 to f192b80 Compare March 1, 2021 05:11

wuchong approved these changes Mar 1, 2021

View reviewed changes

wuchong merged commit cb4bc98 into apache:master Mar 1, 2021

docete deleted the FLINK-18726 branch March 2, 2021 06:59

JingsongLi reviewed Mar 10, 2021

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-18726][table-planner-blink] Support INSERT INTO specific colum… #14977

[FLINK-18726][table-planner-blink] Support INSERT INTO specific colum… #14977

docete commented Feb 22, 2021

flinkbot commented Feb 22, 2021

flinkbot commented Feb 22, 2021 •

edited

Loading

wuchong commented Feb 22, 2021

leonardBang left a comment

leonardBang Feb 22, 2021

docete Feb 23, 2021

docete Feb 24, 2021

sllence commented Feb 24, 2021

docete left a comment

docete Feb 24, 2021

leonardBang left a comment

leonardBang commented Feb 24, 2021 •

edited

Loading

sllence commented Feb 24, 2021

docete commented Feb 25, 2021

wuchong Feb 25, 2021

docete Mar 1, 2021

wuchong left a comment

wuchong commented Mar 1, 2021

JingsongLi Mar 10, 2021

wuchong Mar 10, 2021 •

edited

Loading

JingsongLi Mar 10, 2021

JingsongLi Mar 10, 2021

	expectedEx.expectMessage("Column 'a' has no default value and does not allow NULLs")
	expectedEx.expectMessage("Column 'a' does not allow NULL value, thus it can not be used in partial insert")

[FLINK-18726][table-planner-blink] Support INSERT INTO specific colum… #14977

[FLINK-18726][table-planner-blink] Support INSERT INTO specific colum… #14977

Conversation

docete commented Feb 22, 2021

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 Feb 22, 2021

Automated Checks

Review Progress

flinkbot commented Feb 22, 2021 • edited Loading

CI report:

wuchong commented Feb 22, 2021

leonardBang 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

sllence commented Feb 24, 2021

docete left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leonardBang left a comment

Choose a reason for hiding this comment

leonardBang commented Feb 24, 2021 • edited Loading

sllence commented Feb 24, 2021

docete commented Feb 25, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wuchong left a comment

Choose a reason for hiding this comment

wuchong commented Mar 1, 2021

Choose a reason for hiding this comment

wuchong Mar 10, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flinkbot commented Feb 22, 2021 •

edited

Loading

leonardBang commented Feb 24, 2021 •

edited

Loading

wuchong Mar 10, 2021 •

edited

Loading