You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "luoyuxia (via GitHub)" <gi...@apache.org> on 2023/05/05 11:43:19 UTC
[GitHub] [flink] luoyuxia opened a new pull request, #22525: [FLINK-32001][table] Row-level update should support returning partial columns
luoyuxia opened a new pull request, #22525:
URL: https://github.com/apache/flink/pull/22525
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
*(For example: This pull request makes task deployment go through the blob server, rather than through RPC. That way we avoid re-transferring them on each deployment (during recovery).)*
## Brief change log
*(for example:)*
- *The TaskInfo is stored in the blob store on job creation time as a persistent artifact*
- *Deployments RPC transmits only the blob storage reference*
- *TaskManagers retrieve the TaskInfo from the blob cache*
## Verifying this change
Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (100MB)*
- *Extended integration test for recovery after master (JobManager) failure*
- *Added test that validates that TaskInfo is transferred only once across recoveries*
- *Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.*
## 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #22525:
URL: https://github.com/apache/flink/pull/22525#issuecomment-1550611212
@fsk119 Thanks for reviewing again. Address your comments in [19c680d|https://github.com/apache/flink/pull/22525/commits/19c680d1896c1424efd423645b33d2f179ae0304\
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22525:
URL: https://github.com/apache/flink/pull/22525#discussion_r1195861072
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/sink/RowLevelDeleteSpec.java:
##########
@@ -78,6 +87,11 @@ public SupportsRowLevelDelete.RowLevelDeleteMode getRowLevelDeleteMode() {
return rowLevelDeleteMode;
}
+ @Nonnull
Review Comment:
Make sense to me. I'll remove it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia merged pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #22525:
URL: https://github.com/apache/flink/pull/22525
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #22525:
URL: https://github.com/apache/flink/pull/22525#issuecomment-1551034679
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22525:
URL: https://github.com/apache/flink/pull/22525#discussion_r1195872817
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java:
##########
@@ -647,15 +649,51 @@ private InternalTypeInfo<RowData> getInputTypeInfo() {
}
private int[] getPrimaryKeyIndices(RowType sinkRowType, ResolvedSchema schema) {
- return schema.getPrimaryKey()
- .map(k -> k.getColumns().stream().mapToInt(sinkRowType::getFieldIndex).toArray())
- .orElse(new int[0]);
+ if (schema.getPrimaryKey().isPresent()) {
+ UniqueConstraint uniqueConstraint = schema.getPrimaryKey().get();
+ int[] primaryKeyIndices = new int[uniqueConstraint.getColumns().size()];
+ // sinkRowType may not contain full primary keys in case of row-level update or delete.
+ // in such case, return an empty array
+ for (int i = 0; i < uniqueConstraint.getColumns().size(); i++) {
+ int fieldIndex = sinkRowType.getFieldIndex(uniqueConstraint.getColumns().get(i));
+ if (fieldIndex == -1) {
+ return new int[0];
+ }
Review Comment:
We use primary keys as shuffle keys, but they still are primary keys which maybe used for other purposes in the `CommonExecSink`. If not all primary keys are available, we should consider it as no primary keys which is the semantic of primary keys.
Btw, I have moved such logic to `BatchExecSink` to make every clear since only update statement which is only supported in batch needs to consider it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a diff in pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #22525:
URL: https://github.com/apache/flink/pull/22525#discussion_r1194926178
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java:
##########
@@ -647,15 +649,51 @@ private InternalTypeInfo<RowData> getInputTypeInfo() {
}
private int[] getPrimaryKeyIndices(RowType sinkRowType, ResolvedSchema schema) {
- return schema.getPrimaryKey()
- .map(k -> k.getColumns().stream().mapToInt(sinkRowType::getFieldIndex).toArray())
- .orElse(new int[0]);
+ if (schema.getPrimaryKey().isPresent()) {
+ UniqueConstraint uniqueConstraint = schema.getPrimaryKey().get();
+ int[] primaryKeyIndices = new int[uniqueConstraint.getColumns().size()];
+ // sinkRowType may not contain full primary keys in case of row-level update or delete.
+ // in such case, return an empty array
+ for (int i = 0; i < uniqueConstraint.getColumns().size(); i++) {
+ int fieldIndex = sinkRowType.getFieldIndex(uniqueConstraint.getColumns().get(i));
+ if (fieldIndex == -1) {
+ return new int[0];
+ }
Review Comment:
I think we should try our best to validate all avaliable pk columns.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/sink/RowLevelDeleteSpec.java:
##########
@@ -78,6 +87,11 @@ public SupportsRowLevelDelete.RowLevelDeleteMode getRowLevelDeleteMode() {
return rowLevelDeleteMode;
}
+ @Nonnull
+ public int[] getRequiredPhysicalColumnIndices() {
+ return requiredPhysicalColumnIndices;
+ }
+
@Override
public boolean equals(Object o) {
Review Comment:
Don't forget to modify the equals method.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/sink/RowLevelDeleteSpec.java:
##########
@@ -38,27 +38,36 @@
/**
* A sub-class of {@link SinkAbilitySpec} that can not only serialize/deserialize the row-level
- * delete mode to/from JSON, but also can delete existing data for {@link
- * org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete}.
+ * delete mode & required physical column indices to/from JSON, but also can delete existing data
+ * for {@link org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete}.
*/
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonTypeName("RowLevelDelete")
public class RowLevelDeleteSpec implements SinkAbilitySpec {
public static final String FIELD_NAME_ROW_LEVEL_DELETE_MODE = "rowLevelDeleteMode";
+ public static final String FIELD_NAME_REQUIRED_PHYSICAL_COLUMN_INDICES =
+ "requiredPhysicalColumnIndices";
Review Comment:
`ProjectPushDownSpec` also contains an array to mark the projection. I think `requiredPhysicalColumn` is enough.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java:
##########
@@ -647,15 +649,51 @@ private InternalTypeInfo<RowData> getInputTypeInfo() {
}
private int[] getPrimaryKeyIndices(RowType sinkRowType, ResolvedSchema schema) {
- return schema.getPrimaryKey()
- .map(k -> k.getColumns().stream().mapToInt(sinkRowType::getFieldIndex).toArray())
- .orElse(new int[0]);
+ if (schema.getPrimaryKey().isPresent()) {
+ UniqueConstraint uniqueConstraint = schema.getPrimaryKey().get();
+ int[] primaryKeyIndices = new int[uniqueConstraint.getColumns().size()];
+ // sinkRowType may not contain full primary keys in case of row-level update or delete.
+ // in such case, return an empty array
+ for (int i = 0; i < uniqueConstraint.getColumns().size(); i++) {
+ int fieldIndex = sinkRowType.getFieldIndex(uniqueConstraint.getColumns().get(i));
+ if (fieldIndex == -1) {
+ return new int[0];
+ }
+ primaryKeyIndices[i] = fieldIndex;
+ }
+ return primaryKeyIndices;
+ } else {
+ return new int[0];
+ }
}
private RowType getPhysicalRowType(ResolvedSchema schema) {
+ // row-level modification may only write partial columns
+ if (tableSinkSpec.getSinkAbilities() != null) {
+ for (SinkAbilitySpec sinkAbilitySpec : tableSinkSpec.getSinkAbilities()) {
+ if (sinkAbilitySpec instanceof RowLevelUpdateSpec) {
+ RowLevelUpdateSpec rowLevelUpdateSpec = (RowLevelUpdateSpec) sinkAbilitySpec;
+ return getPhysicalRowType(schema, rowLevelUpdateSpec.getRequireColumnIndices());
+ } else if (sinkAbilitySpec instanceof RowLevelDeleteSpec) {
+ RowLevelDeleteSpec rowLevelDeleteSpec = (RowLevelDeleteSpec) sinkAbilitySpec;
+ return getPhysicalRowType(
+ schema, rowLevelDeleteSpec.getRequiredPhysicalColumnIndices());
Review Comment:
I just wonder whether it's better to introduce a method like `Optional<RowType> getConsumedType` for this sink ability spec. Actually, we already have `getProducedType` in the source ability spec.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/sink/RowLevelDeleteSpec.java:
##########
@@ -78,6 +87,11 @@ public SupportsRowLevelDelete.RowLevelDeleteMode getRowLevelDeleteMode() {
return rowLevelDeleteMode;
}
+ @Nonnull
Review Comment:
remove the annotation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #22525:
URL: https://github.com/apache/flink/pull/22525#issuecomment-1551065434
@lincoln-lil @fsk119 Thanks for your review. I have addressed your comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #22525:
URL: https://github.com/apache/flink/pull/22525#issuecomment-1549692540
@fsk119 Thanks for your review. Address your comments in [9b4b58f](https://github.com/apache/flink/commit/9b4b58f2b00ad8705728120a191dec04fec13db6)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22525:
URL: https://github.com/apache/flink/pull/22525#discussion_r1195172283
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/sink/RowLevelDeleteSpec.java:
##########
@@ -38,27 +38,36 @@
/**
* A sub-class of {@link SinkAbilitySpec} that can not only serialize/deserialize the row-level
- * delete mode to/from JSON, but also can delete existing data for {@link
- * org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete}.
+ * delete mode & required physical column indices to/from JSON, but also can delete existing data
+ * for {@link org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete}.
*/
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonTypeName("RowLevelDelete")
public class RowLevelDeleteSpec implements SinkAbilitySpec {
public static final String FIELD_NAME_ROW_LEVEL_DELETE_MODE = "rowLevelDeleteMode";
+ public static final String FIELD_NAME_REQUIRED_PHYSICAL_COLUMN_INDICES =
+ "requiredPhysicalColumnIndices";
Review Comment:
What do you mean by saying `requiredPhysicalColumn` is enough? AFAIC, `ProjectPushDownSpec` also contains an array to mark the projection, so we can also make `RowLevelDeleteSpec` contains an array to mark the required physical columns.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22525:
URL: https://github.com/apache/flink/pull/22525#discussion_r1195167785
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java:
##########
@@ -647,15 +649,51 @@ private InternalTypeInfo<RowData> getInputTypeInfo() {
}
private int[] getPrimaryKeyIndices(RowType sinkRowType, ResolvedSchema schema) {
- return schema.getPrimaryKey()
- .map(k -> k.getColumns().stream().mapToInt(sinkRowType::getFieldIndex).toArray())
- .orElse(new int[0]);
+ if (schema.getPrimaryKey().isPresent()) {
+ UniqueConstraint uniqueConstraint = schema.getPrimaryKey().get();
+ int[] primaryKeyIndices = new int[uniqueConstraint.getColumns().size()];
+ // sinkRowType may not contain full primary keys in case of row-level update or delete.
+ // in such case, return an empty array
+ for (int i = 0; i < uniqueConstraint.getColumns().size(); i++) {
+ int fieldIndex = sinkRowType.getFieldIndex(uniqueConstraint.getColumns().get(i));
+ if (fieldIndex == -1) {
+ return new int[0];
+ }
Review Comment:
In here, we do validate all avaliable pk columns.
The logic is try to find all the pk columns from the `ResolvedSchema`, but if for one pk column, we find we can't find from the `ResolvedSchema`, that means the pk columns miss some columns in which case we will consider it as no pk columns.
Note: this case only happens in update statement, the required columns returned by connector don't contain all pk columns.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #22525:
URL: https://github.com/apache/flink/pull/22525#issuecomment-1536140236
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "78156c4aa2cadcbe6783e6d70901a9a9ec2601f1",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "78156c4aa2cadcbe6783e6d70901a9a9ec2601f1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 78156c4aa2cadcbe6783e6d70901a9a9ec2601f1 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a diff in pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #22525:
URL: https://github.com/apache/flink/pull/22525#discussion_r1195847470
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/sink/RowLevelDeleteSpec.java:
##########
@@ -78,6 +87,11 @@ public SupportsRowLevelDelete.RowLevelDeleteMode getRowLevelDeleteMode() {
return rowLevelDeleteMode;
}
+ @Nonnull
Review Comment:
Actually in the code style, it suggests it's not a necessary behaviour to mark the field not null. BTW, it's very verbose if we mark every method's return type not null[1].
[1] https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#nullability-of-the-mutable-parts
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java:
##########
@@ -647,15 +649,51 @@ private InternalTypeInfo<RowData> getInputTypeInfo() {
}
private int[] getPrimaryKeyIndices(RowType sinkRowType, ResolvedSchema schema) {
- return schema.getPrimaryKey()
- .map(k -> k.getColumns().stream().mapToInt(sinkRowType::getFieldIndex).toArray())
- .orElse(new int[0]);
+ if (schema.getPrimaryKey().isPresent()) {
+ UniqueConstraint uniqueConstraint = schema.getPrimaryKey().get();
+ int[] primaryKeyIndices = new int[uniqueConstraint.getColumns().size()];
+ // sinkRowType may not contain full primary keys in case of row-level update or delete.
+ // in such case, return an empty array
+ for (int i = 0; i < uniqueConstraint.getColumns().size(); i++) {
+ int fieldIndex = sinkRowType.getFieldIndex(uniqueConstraint.getColumns().get(i));
+ if (fieldIndex == -1) {
+ return new int[0];
+ }
+ primaryKeyIndices[i] = fieldIndex;
+ }
+ return primaryKeyIndices;
+ } else {
+ return new int[0];
+ }
}
private RowType getPhysicalRowType(ResolvedSchema schema) {
+ // row-level modification may only write partial columns
+ if (tableSinkSpec.getSinkAbilities() != null) {
+ for (SinkAbilitySpec sinkAbilitySpec : tableSinkSpec.getSinkAbilities()) {
+ if (sinkAbilitySpec instanceof RowLevelUpdateSpec) {
+ RowLevelUpdateSpec rowLevelUpdateSpec = (RowLevelUpdateSpec) sinkAbilitySpec;
+ return getPhysicalRowType(schema, rowLevelUpdateSpec.getRequireColumnIndices());
+ } else if (sinkAbilitySpec instanceof RowLevelDeleteSpec) {
+ RowLevelDeleteSpec rowLevelDeleteSpec = (RowLevelDeleteSpec) sinkAbilitySpec;
+ return getPhysicalRowType(
+ schema, rowLevelDeleteSpec.getRequiredPhysicalColumnIndices());
Review Comment:
Fine. We can modify this if we needed in the future.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java:
##########
@@ -647,15 +649,51 @@ private InternalTypeInfo<RowData> getInputTypeInfo() {
}
private int[] getPrimaryKeyIndices(RowType sinkRowType, ResolvedSchema schema) {
- return schema.getPrimaryKey()
- .map(k -> k.getColumns().stream().mapToInt(sinkRowType::getFieldIndex).toArray())
- .orElse(new int[0]);
+ if (schema.getPrimaryKey().isPresent()) {
+ UniqueConstraint uniqueConstraint = schema.getPrimaryKey().get();
+ int[] primaryKeyIndices = new int[uniqueConstraint.getColumns().size()];
+ // sinkRowType may not contain full primary keys in case of row-level update or delete.
+ // in such case, return an empty array
+ for (int i = 0; i < uniqueConstraint.getColumns().size(); i++) {
+ int fieldIndex = sinkRowType.getFieldIndex(uniqueConstraint.getColumns().get(i));
+ if (fieldIndex == -1) {
+ return new int[0];
+ }
Review Comment:
I mean why we don't return all available positions here?
```
uniqueConstraint.getColumns().stream()
.mapToInt(sinkRowType::getFieldIndex)
.filter(i -> i == -1)
.toArray();
```
I find the indcies is used as shuffle key, in the update statement we don't need shuffle?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/sink/RowLevelDeleteSpec.java:
##########
@@ -38,27 +38,36 @@
/**
* A sub-class of {@link SinkAbilitySpec} that can not only serialize/deserialize the row-level
- * delete mode to/from JSON, but also can delete existing data for {@link
- * org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete}.
+ * delete mode & required physical column indices to/from JSON, but also can delete existing data
+ * for {@link org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete}.
*/
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonTypeName("RowLevelDelete")
public class RowLevelDeleteSpec implements SinkAbilitySpec {
public static final String FIELD_NAME_ROW_LEVEL_DELETE_MODE = "rowLevelDeleteMode";
+ public static final String FIELD_NAME_REQUIRED_PHYSICAL_COLUMN_INDICES =
+ "requiredPhysicalColumnIndices";
Review Comment:
oh. I mean it's better to rename to `requiredPhysicalColumn` to align the name behaviour. It's fine to use `requiredPhysicalColumnIndices`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lincoln-lil commented on a diff in pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "lincoln-lil (via GitHub)" <gi...@apache.org>.
lincoln-lil commented on code in PR #22525:
URL: https://github.com/apache/flink/pull/22525#discussion_r1196190483
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/DeleteTableITCase.java:
##########
@@ -138,7 +181,7 @@ public void testMixDelete() throws Exception {
@Test
public void testStatementSetContainDeleteAndInsert() throws Exception {
tEnv().executeSql(
- "CREATE TABLE t (a int, b string, c double) WITH"
+ "CREATE TABLE t (a int , b string, c double) WITH"
Review Comment:
nit: rm extra space
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #22525:
URL: https://github.com/apache/flink/pull/22525#issuecomment-1551064562
> @luoyuxia thanks for fixing this! Overall looks good to me, and it is recommended to add some more cases to cover partition, metadata(both virtual and non-virtual) and computed columns.
Agree. I create a separate Jira FLINK-32117 to track it as it seems a separate issue to me. I expect to finish it in another jira.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22525:
URL: https://github.com/apache/flink/pull/22525#discussion_r1195177400
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java:
##########
@@ -647,15 +649,51 @@ private InternalTypeInfo<RowData> getInputTypeInfo() {
}
private int[] getPrimaryKeyIndices(RowType sinkRowType, ResolvedSchema schema) {
- return schema.getPrimaryKey()
- .map(k -> k.getColumns().stream().mapToInt(sinkRowType::getFieldIndex).toArray())
- .orElse(new int[0]);
+ if (schema.getPrimaryKey().isPresent()) {
+ UniqueConstraint uniqueConstraint = schema.getPrimaryKey().get();
+ int[] primaryKeyIndices = new int[uniqueConstraint.getColumns().size()];
+ // sinkRowType may not contain full primary keys in case of row-level update or delete.
+ // in such case, return an empty array
+ for (int i = 0; i < uniqueConstraint.getColumns().size(); i++) {
+ int fieldIndex = sinkRowType.getFieldIndex(uniqueConstraint.getColumns().get(i));
+ if (fieldIndex == -1) {
+ return new int[0];
+ }
+ primaryKeyIndices[i] = fieldIndex;
+ }
+ return primaryKeyIndices;
+ } else {
+ return new int[0];
+ }
}
private RowType getPhysicalRowType(ResolvedSchema schema) {
+ // row-level modification may only write partial columns
+ if (tableSinkSpec.getSinkAbilities() != null) {
+ for (SinkAbilitySpec sinkAbilitySpec : tableSinkSpec.getSinkAbilities()) {
+ if (sinkAbilitySpec instanceof RowLevelUpdateSpec) {
+ RowLevelUpdateSpec rowLevelUpdateSpec = (RowLevelUpdateSpec) sinkAbilitySpec;
+ return getPhysicalRowType(schema, rowLevelUpdateSpec.getRequireColumnIndices());
+ } else if (sinkAbilitySpec instanceof RowLevelDeleteSpec) {
+ RowLevelDeleteSpec rowLevelDeleteSpec = (RowLevelDeleteSpec) sinkAbilitySpec;
+ return getPhysicalRowType(
+ schema, rowLevelDeleteSpec.getRequiredPhysicalColumnIndices());
Review Comment:
There're many places use `getProducedType` in the source ability spec, but seems it's the only place to use something like `Optional<RowType> getConsumedType`. I think we can just limit it in here to avoid think too much to early and consider to expose in `SinkabilitySpec` if we found it'll be needed in many places in the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22525:
URL: https://github.com/apache/flink/pull/22525#discussion_r1195167785
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java:
##########
@@ -647,15 +649,51 @@ private InternalTypeInfo<RowData> getInputTypeInfo() {
}
private int[] getPrimaryKeyIndices(RowType sinkRowType, ResolvedSchema schema) {
- return schema.getPrimaryKey()
- .map(k -> k.getColumns().stream().mapToInt(sinkRowType::getFieldIndex).toArray())
- .orElse(new int[0]);
+ if (schema.getPrimaryKey().isPresent()) {
+ UniqueConstraint uniqueConstraint = schema.getPrimaryKey().get();
+ int[] primaryKeyIndices = new int[uniqueConstraint.getColumns().size()];
+ // sinkRowType may not contain full primary keys in case of row-level update or delete.
+ // in such case, return an empty array
+ for (int i = 0; i < uniqueConstraint.getColumns().size(); i++) {
+ int fieldIndex = sinkRowType.getFieldIndex(uniqueConstraint.getColumns().get(i));
+ if (fieldIndex == -1) {
+ return new int[0];
+ }
Review Comment:
In here, we do validate all avaliable pk columns.
The logic is try to find all the pk columns from the `ResolvedSchema`, but if for one pk column, we can't find from the `ResolvedSchema`, that means the pk columns miss some columns in which case we will consider it as no pk columns.
Note: this case only happens in update statement, the required columns returned by connector don't contain all pk columns.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22525:
URL: https://github.com/apache/flink/pull/22525#discussion_r1195170631
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/sink/RowLevelDeleteSpec.java:
##########
@@ -78,6 +87,11 @@ public SupportsRowLevelDelete.RowLevelDeleteMode getRowLevelDeleteMode() {
return rowLevelDeleteMode;
}
+ @Nonnull
Review Comment:
Why remove it? It's no bad to keep it. I think we can keep the annotation just like what we do for the other method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on pull request #22525: [FLINK-32001][table] Row-level update should support returning partial columns
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #22525:
URL: https://github.com/apache/flink/pull/22525#issuecomment-1552275854
Others also encounter the problem as reported in FLINK-32121. The reason is shown as FLINK-32123.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org