[FLINK-30665][table] Planner supports row-level update #21698

luoyuxia · 2023-01-17T06:51:02Z

What is the purpose of the change

To make planner supports row-level update.

Brief change log

Convert the SqUpdate to LogicalTableModify, and then rewrite the LogicalTableModify to a RelNode that insert into the sink with rows
- If the update mode is ALL_ROWS, assuming the filter is a, convert the update expression to IF(a, xx, yyy).
- Get the required columns by calling method SupportsRowLevelUpdate#applyRowLevelUpdate. We may rewrite the
  LogicaTableScan to add extra meta cols to the underlying table. Then create a Project node that only select the required
  columns.
If update mode is UPDATED_ROWS, add a operator to set the RowKind to RowKind#UPDATE_AFTER in CommonExecSink
Move the method SqlToOperationConverterTest#prepareTable to a new class to avoid SqlToOperationConverterTest is too long to fail to pass check-style.

Verifying this change

The test in SqlToOperationConverterTest#testUpdate to verify update statement will be converted to SinkModifyOperation.
The test in RowLevelDeleteUpdate to verify the plan for row-level update.
The IT case in UpdateTableITCase to verify update can be performed normally.

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, Kubernetes/Yarn, 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 · 2023-01-17T06:57:01Z

CI report:

016fd6e Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

luoyuxia · 2023-01-28T09:29:40Z

@lincoln-lil The pr is ready to review. Could you please help review when you're free?

… a new separate SqlToOperationConverterTestUtils to avoid SqlToOperationConverterTest is too long to make check-style fail

lincoln-lil

@luoyuxia thanks for contributing this! Looks good to me overall, I've left some comments.

lincoln-lil · 2023-01-29T01:45:41Z

...ner/src/test/java/org/apache/flink/table/planner/utils/SqlToOperationConverterTestUtils.java

+import java.util.Map;
+
+/** Utils for {@link org.apache.flink.table.planner.operations.SqlToOperationConverterTest} . */
+public class SqlToOperationConverterTestUtils {


Since the SqlToOperationConverterTest has exceeded the maximum row limit, we might as well split it by sql type, such as ddl, dml and other command, extract a SqlToOperationConverterTestBase to keep all the helper methods, and split the test into three pieces: SqlDDLToOperationConverterTest, SqlDMLToOperationConverterTest, and SqlCommandToOperationConverterTest , WDYT?

A good suggestion.

lincoln-lil · 2023-01-29T02:04:21Z

...planner/src/main/java/org/apache/flink/table/planner/operations/SqlToOperationConverter.java

+        RelRoot updateRelational = flinkPlanner.rel(sqlUpdate);
+        // get target sink table
+        LogicalTableModify tableModify = (LogicalTableModify) updateRelational.rel;
+        List<String> targetTablePath = tableModify.getTable().getQualifiedName();


nit: use UnresolvedIdentifier directly to keep consistent with the delete operation

lincoln-lil · 2023-01-29T02:09:07Z

...ner/src/main/java/org/apache/flink/table/planner/plan/abilities/sink/RowLevelUpdateSpec.java

+
+/**
+ * A sub-class of {@link SinkAbilitySpec} that can not only serialize/deserialize the row-level
+ * update mode & updated columns to/from JSON, but also can update existing data for {@link


'update mode & updated columns' -> 'update mode & columns'

lincoln-lil · 2023-01-29T02:16:52Z

...-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java

+        List<Column> updatedColumns = getUpdatedColumns(tableModify, resolvedSchema);
+        SupportsRowLevelUpdate.RowLevelUpdateInfo updateInfo =
+                supportsRowLevelUpdate.applyRowLevelUpdate(
+                        updatedColumns, RowLevelModificationContextUtils.getScanContext());


nit: add a local variable for RowLevelModificationContextUtils.getScanContext()

lincoln-lil · 2023-01-29T02:39:53Z

...ner/src/test/java/org/apache/flink/table/planner/factories/TestUpdateDeleteTableFactory.java

+                    .withDescription("The name for the required columns in row-level update");
+
+    private static final ConfigOption<Boolean> ONLY_REQUIRE_UPDATED_COLUMNS_FOR_UPDATE =
+            ConfigOptions.key("only_require_updated_columns_for_update")


'only_require_updated_columns_for_update' -> 'only-require-updated-columns-for-update'

lincoln-lil · 2023-01-29T02:42:01Z

...ner/src/test/java/org/apache/flink/table/planner/factories/TestUpdateDeleteTableFactory.java

+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to only require the updated columns for update statement. ");


add additional comments: ', require all columns by default'

lincoln-lil · 2023-01-29T02:46:47Z

...-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/RowLevelUpdateTest.java

+    }
+
+    @Test
+    public void testUpdateWithOnlyRequireUpdatedCols() {


add a new case which not contains all columns in update clause, e.g., a source table with more columns here 'TABLE t (f0, f1,..., a int, b string, c double, ...)'

…nts for row-level update

luoyuxia · 2023-01-29T09:42:42Z

@lincoln-lil Thanks for reviewing. I append two commits to fix your comments. The first one is to address the comments for row-level update, the second one is to split the test SqlToOperationConverterTest.

…three different test classes to avoid it exceed the maximum row limit

lincoln-lil

@luoyuxia thanks for your update! I have a minor comment for the test splitting: although there is no strict guideline on how to classify various sql commands, it is clearly inappropriate to classify commands such as show, load, use, etc. into ddl or dml, should we adjust it to refer to mysql's definition of ddl, dml, and put the remaining into sqlOtherOperation (probably called sqlCommand is also easy to cause ambiguity).
Btw, since 'begin statement set' is used to assist multiple dml statements, we can also put it into dml test, WDYT?

luoyuxia · 2023-01-30T03:34:39Z

@luoyuxia thanks for your update! I have a minor comment for the test splitting: although there is no strict guideline on how to classify various sql commands, it is clearly inappropriate to classify commands such as show, load, use, etc. into ddl or dml, should we adjust it to refer to mysql's definition of ddl, dml, and put the remaining into sqlOtherOperation (probably called sqlCommand is also easy to cause ambiguity).

Agree to it. The reason I put show/use into ddl is from the Hive's language manual ddl wiki. But now, I think the hive's is not correct and suitable for all systems. So i like the idea that put them into sqlOtherOperation.

Btw, since 'begin statement set' is used to assist multiple dml statements, we can also put it into dml test, WDYT?

Agree to it

luoyuxia · 2023-01-30T03:52:02Z

@lincoln-lil Thanks for reviewing. Now I move the test for the statements that neither belong to DDL nor DML to SqlOtherOperationConverterTest

lincoln-lil · 2023-01-30T06:59:06Z

@flinkbot run azure

This closes apache#21698

lincoln-lil

LGTM

This closes apache#21698

flinkbot added the component=TableSQL/API label Jan 17, 2023

luoyuxia force-pushed the FLINK-30665 branch 5 times, most recently from a4db6c0 to a3f1c61 Compare January 28, 2023 09:03

luoyuxia force-pushed the FLINK-30665 branch 2 times, most recently from b25d2be to 3dd9fa1 Compare January 28, 2023 12:16

luoyuxia added 2 commits January 29, 2023 10:18

[FLINK-30665][table] Planner supports row-level update

270473c

[FLINK-30665][table] move SqlToOperationConverterTest#prepareTable to…

448a52e

… a new separate SqlToOperationConverterTestUtils to avoid SqlToOperationConverterTest is too long to make check-style fail

luoyuxia force-pushed the FLINK-30665 branch from 3dd9fa1 to 448a52e Compare January 29, 2023 02:18

lincoln-lil reviewed Jan 29, 2023

View reviewed changes

[FLINK-30665][table] Planner supports row-level update: address comme…

c441676

…nts for row-level update

[FLINK-30665][table] Split the test SqlToOperationConverterTest into …

016fd6e

…three different test classes to avoid it exceed the maximum row limit

luoyuxia force-pushed the FLINK-30665 branch from 80db065 to 016fd6e Compare January 29, 2023 10:27

lincoln-lil reviewed Jan 30, 2023

View reviewed changes

[FLINK-30665][table] move some test to SqlOtherOperationConverterTest

a03cec4

luoyuxia force-pushed the FLINK-30665 branch from 0e4c9a8 to a03cec4 Compare January 30, 2023 03:50

lincoln-lil pushed a commit to lincoln-lil/flink that referenced this pull request Jan 30, 2023

[FLINK-30665][table] Planner supports row-level update

fd66e85

This closes apache#21698

lincoln-lil approved these changes Jan 30, 2023

View reviewed changes

lincoln-lil closed this in 5afc301 Jan 30, 2023

chucheng92 pushed a commit to chucheng92/flink that referenced this pull request Feb 3, 2023

[FLINK-30665][table] Planner supports row-level update

4830204

This closes apache#21698

akkinenivijay pushed a commit to krisnaru/flink that referenced this pull request Feb 11, 2023

[FLINK-30665][table] Planner supports row-level update

c027ca8

This closes apache#21698

[FLINK-30665][table] Planner supports row-level update #21698

[FLINK-30665][table] Planner supports row-level update #21698

Uh oh!

Conversation

luoyuxia commented Jan 17, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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 Jan 17, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

luoyuxia commented Jan 28, 2023

Uh oh!

lincoln-lil 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!

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!

luoyuxia commented Jan 29, 2023

Uh oh!

lincoln-lil left a comment

Choose a reason for hiding this comment

Uh oh!

luoyuxia commented Jan 30, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

luoyuxia commented Jan 30, 2023

Uh oh!

lincoln-lil commented Jan 30, 2023

Uh oh!

lincoln-lil left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

luoyuxia commented Jan 17, 2023 •

edited

Loading

flinkbot commented Jan 17, 2023 •

edited

Loading

luoyuxia commented Jan 30, 2023 •

edited

Loading