You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "manuzhang (via GitHub)" <gi...@apache.org> on 2023/08/01 05:52:14 UTC
[GitHub] [iceberg] manuzhang opened a new pull request, #8195: Flink 1.17: Support specifying equality columns with write options
manuzhang opened a new pull request, #8195:
URL: https://github.com/apache/iceberg/pull/8195
Currently, equality columns are parsed from primary key specified in Flink create table statement. However, an Iceberg table might have been created without primary key, but we still want to upsert with Flink. In this case, we need another way to specify equality columns. This PR attempts to solve it with write options.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.17: Support specifying equality columns with write options [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #8195:
URL: https://github.com/apache/iceberg/pull/8195#issuecomment-1858158123
> However, there's no primary key in Spark or Trino.
I think it should be possible to provide the identity fields there too.
The java API provides the possibility through the `Schema`
https://github.com/apache/iceberg/blob/7c3c34897a2e452311f085af67d6f546ba56e44e/api/src/main/java/org/apache/iceberg/Schema.java#L82
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.17: Support specifying equality columns with write options [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #8195:
URL: https://github.com/apache/iceberg/pull/8195#issuecomment-1855685224
> Currently, equality columns are parsed from primary key specified in Flink create table statement. However, an Iceberg table might have been created without primary key, but we still want to upsert with Flink. In this case, we need another way to specify equality columns. This PR attempts to solve it with write options.
Wouldn't be a better solution to just update the table definition? If the given columns are really primary keys, then the table would benefit from the better specification. If it is not really a primary key, then there could be data corruption when using it as a primary key for upserts.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.17: Support specifying equality columns with write options [iceberg]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on PR #8195:
URL: https://github.com/apache/iceberg/pull/8195#issuecomment-1892165875
Closing this since Spark table can support Flink SQL upsert operation with identifier fields.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.17: Support specifying equality columns with write options [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #8195:
URL: https://github.com/apache/iceberg/pull/8195#issuecomment-1855873133
@manuzhang: The main point of Iceberg is that we have an engine independent table format. The columns are either primary keys, or not. If they are primary keys, they should be primary keys for Spark, Flink, Hive or Trino. So IMHO the table should define which columns are primary keys.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #8195: Flink 1.17: Support specifying equality columns with write options
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #8195:
URL: https://github.com/apache/iceberg/pull/8195#discussion_r1302361554
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java:
##########
@@ -379,12 +381,11 @@ private String operatorName(String suffix) {
}
@VisibleForTesting
- List<Integer> checkAndGetEqualityFieldIds() {
+ List<Integer> checkAndGetEqualityFieldIds(List<String> equalityColumns) {
Review Comment:
We have assigned `this.equalityFieldColumns`, so is there a need to change here?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.17: Support specifying equality columns with write options [iceberg]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on PR #8195:
URL: https://github.com/apache/iceberg/pull/8195#issuecomment-1855777652
@pvary I've updated my description. Our use case is that the table is created in Spark while data are sinked from Flink
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] manuzhang commented on pull request #8195: Flink 1.17: Support specifying equality columns with write options
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on PR #8195:
URL: https://github.com/apache/iceberg/pull/8195#issuecomment-1667367689
@hililiwei please take another look
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] manuzhang commented on a diff in pull request #8195: Flink 1.17: Support specifying equality columns with write options
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on code in PR #8195:
URL: https://github.com/apache/iceberg/pull/8195#discussion_r1304045323
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java:
##########
@@ -379,12 +381,11 @@ private String operatorName(String suffix) {
}
@VisibleForTesting
- List<Integer> checkAndGetEqualityFieldIds() {
+ List<Integer> checkAndGetEqualityFieldIds(List<String> equalityColumns) {
Review Comment:
It's for testing. Please check the changes in `[TestFlinkIcebergSinkV2.java](https://github.com/apache/iceberg/pull/8195/files#diff-560cafe3f640ca5cdff37ff1af2fe15d7e87dd644ca2669bf742c2999628bd5d)`
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java:
##########
@@ -379,12 +381,11 @@ private String operatorName(String suffix) {
}
@VisibleForTesting
- List<Integer> checkAndGetEqualityFieldIds() {
+ List<Integer> checkAndGetEqualityFieldIds(List<String> equalityColumns) {
Review Comment:
It's for testing. Please check the changes in [TestFlinkIcebergSinkV2.java](https://github.com/apache/iceberg/pull/8195/files#diff-560cafe3f640ca5cdff37ff1af2fe15d7e87dd644ca2669bf742c2999628bd5d)
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.17: Support specifying equality columns with write options [iceberg]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on PR #8195:
URL: https://github.com/apache/iceberg/pull/8195#issuecomment-1858100586
However, there's no primary key in Spark or Trino.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.17: Support specifying equality columns with write options [iceberg]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on PR #8195:
URL: https://github.com/apache/iceberg/pull/8195#issuecomment-1855640410
@stevenzwu @pvary please help review
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 1.17: Support specifying equality columns with write options [iceberg]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang closed pull request #8195: Flink 1.17: Support specifying equality columns with write options
URL: https://github.com/apache/iceberg/pull/8195
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] manuzhang commented on pull request #8195: Flink 1.17: Support specifying equality columns with write options
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on PR #8195:
URL: https://github.com/apache/iceberg/pull/8195#issuecomment-1659716109
@hililiwei @stevenzwu please help review.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #8195: Flink 1.17: Support specifying equality columns with write options
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #8195:
URL: https://github.com/apache/iceberg/pull/8195#discussion_r1283145459
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java:
##########
@@ -71,7 +72,10 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
"Unbounded data stream doesn't support overwrite operation.");
List<String> equalityColumns =
- tableSchema.getPrimaryKey().map(UniqueConstraint::getColumns).orElseGet(ImmutableList::of);
+ tableSchema
+ .getPrimaryKey()
+ .map(UniqueConstraint::getColumns)
+ .orElseGet(() -> getEqualityColumnsFromProps(writeProps));
Review Comment:
We don't parse options this way.
You should refer to FlinkWriteOptions. Put`equlality columns` in `writeOptions`( `FlinkSink.Builder#equalityFieldColumns`) like we did in `FlinkSink.Builder#upsert`.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org