mxm commented on code in PR #18668:
URL: https://github.com/apache/iceberg/pull/18668#discussion_r4403270605
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java:
##########
@@ -117,23 +122,44 @@ private KeySelector<RowData, Integer> getKeySelector(
PartitionSpec spec,
DistributionMode mode,
Set<String> equalityFields,
- int writeParallelism) {
+ int writeParallelism,
+ Set<String> distributionFields) {
LOG.debug(
"Creating new KeySelector for table '{}' with distribution mode '{}'",
tableName, mode);
+
switch (mode) {
case NONE:
+ if (!distributionFields.isEmpty()) {
+ LOG.warn(
+ "{}: Distribution fields {} are ignored when distribution mode
is set to NONE",
+ tableName,
+ distributionFields);
+ }
+
if (equalityFields.isEmpty()) {
return tableKeySelector(tableName, writeParallelism,
maxWriteParallelism);
} else {
LOG.info(
"{}: Distribute rows by equality fields, because there are
equality fields set",
tableName);
- return equalityFieldKeySelector(
- tableName, schema, equalityFields, writeParallelism,
maxWriteParallelism);
+ return fieldKeySelector(
+ tableName,
+ schema,
+ DynamicSinkUtil.getEqualityFieldIds(equalityFields, schema),
+ writeParallelism,
+ maxWriteParallelism);
}
case HASH:
- if (equalityFields.isEmpty()) {
+ if (!distributionFields.isEmpty()) {
Review Comment:
When equality fields are set (or the schema identifier fields), this can
route rows with the same key to different writers, i.e. if the distribution
fields aren't part of the key. In upsert mode, both writers then write an
equality delete plus an insert for that key in the same checkpoint, which is
not correct.
Should we require the distribution fields to be a subset of the equality
fields, similar to the [partition
fields](https://github.com/apache/iceberg/blob/8c41e9f7f283bfa18a7c67c7f56f65b6df6538e3/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java#L193)?
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java:
##########
@@ -117,23 +122,44 @@ private KeySelector<RowData, Integer> getKeySelector(
PartitionSpec spec,
DistributionMode mode,
Set<String> equalityFields,
- int writeParallelism) {
+ int writeParallelism,
+ Set<String> distributionFields) {
LOG.debug(
"Creating new KeySelector for table '{}' with distribution mode '{}'",
tableName, mode);
+
switch (mode) {
case NONE:
+ if (!distributionFields.isEmpty()) {
+ LOG.warn(
+ "{}: Distribution fields {} are ignored when distribution mode
is set to NONE",
+ tableName,
+ distributionFields);
+ }
Review Comment:
I think we should not ignore the distribution fields, similarly how
equalityFields are handled in DistributionMode NONE.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java:
##########
@@ -632,12 +663,14 @@ void testMultipleTables() throws Exception {
SimpleDataUtil.SCHEMA,
"t1",
SnapshotRef.MAIN_BRANCH,
- PartitionSpec.unpartitioned()),
+ PartitionSpec.unpartitioned(),
+ Collections.singleton("id")),
new DynamicIcebergDataImpl(
SimpleDataUtil.SCHEMA,
"t2",
SnapshotRef.MAIN_BRANCH,
- PartitionSpec.unpartitioned()));
+ PartitionSpec.unpartitioned(),
+ Collections.singleton("id")));
Review Comment:
This test still passes if the distribution fields are ignored: each table
gets a single row, and HASH on an unpartitioned table without equality fields
falls back to round-robin. Could we keep the test as-is and add a separate test
that writes several rows with the same `id` and checks that they end up in a
single data file per commit?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
