sqd commented on code in PR #18675:
URL: https://github.com/apache/iceberg/pull/18675#discussion_r4486309466
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriter.java:
##########
@@ -97,67 +107,116 @@ public void write(DynamicRecordInternal element, Context
context)
element.spec().specId(),
element.upsertMode(),
element.equalityFields()),
- writerKey -> {
- RowDataTaskWriterFactory taskWriterFactory =
- taskWriterFactories.computeIfAbsent(
- writerKey,
- factoryKey -> {
- Table table =
-
catalog.loadTable(TableIdentifier.parse(factoryKey.tableName()));
-
- Set<Integer> equalityFieldIds =
- getEqualityFields(table, element.equalityFields());
- if (element.upsertMode()) {
- Preconditions.checkState(
- !equalityFieldIds.isEmpty(),
- "Equality field columns shouldn't be empty when
configuring to use UPSERT data.");
-
- if (!table.spec().isUnpartitioned()) {
- for (PartitionField partitionField :
table.spec().fields()) {
- Preconditions.checkState(
-
equalityFieldIds.contains(partitionField.sourceId()),
- "In UPSERT mode, partition field '%s' should
be included in equality fields: '%s'",
- partitionField,
- equalityFieldIds);
- }
- }
- }
-
- FlinkWriteConf flinkWriteConf =
- new FlinkWriteConf(table, commonWriteProperties,
flinkConfig);
- Map<String, String> tableWriteProperties =
- SinkUtil.writeProperties(
- flinkWriteConf.dataFileFormat(),
flinkWriteConf, table);
-
- LOG.debug("Creating new writer factory for table
'{}'", table.name());
- return new RowDataTaskWriterFactory(
- () -> table,
- FlinkSchemaUtil.convert(element.schema()),
- flinkWriteConf.targetDataFileSize(),
- flinkWriteConf.dataFileFormat(),
- tableWriteProperties,
- Lists.newArrayList(equalityFieldIds),
- element.upsertMode(),
- element.schema(),
- element.spec());
- });
-
- taskWriterFactory.initialize(subTaskId, attemptId);
- return taskWriterFactory.create();
- })
+ writerKey -> createWriter(writerKey, element))
.write(element.rowData());
metrics.mainMetricsGroup().getNumRecordsSendCounter().inc();
}
+ private TaskWriter<RowData> createWriter(WriteTarget writerKey,
DynamicRecordInternal element) {
+ RowDataTaskWriterFactory taskWriterFactory =
+ taskWriterFactories.computeIfAbsent(
+ writerKey,
+ factoryKey -> {
+ Table table =
catalog.loadTable(TableIdentifier.parse(factoryKey.tableName()));
+
+ Set<Integer> equalityFieldIds = getEqualityFields(table,
element.equalityFields());
+ if (element.upsertMode()) {
+ Preconditions.checkState(
+ !equalityFieldIds.isEmpty(),
+ "Equality field columns shouldn't be empty when
configuring to use UPSERT data.");
+
+ if (!table.spec().isUnpartitioned()) {
+ for (PartitionField partitionField : table.spec().fields()) {
+ Preconditions.checkState(
+ equalityFieldIds.contains(partitionField.sourceId()),
+ "In UPSERT mode, partition field '%s' should be
included in equality fields: '%s'",
+ partitionField,
+ equalityFieldIds);
+ }
+ }
+ }
+
+ FlinkWriteConf flinkWriteConf =
+ new FlinkWriteConf(table, commonWriteProperties,
flinkConfig);
+ Map<String, String> tableWriteProperties =
+ SinkUtil.writeProperties(flinkWriteConf.dataFileFormat(),
flinkWriteConf, table);
+
+ WriteTarget trackerScope = factoryKey.withoutSchemaId();
+ LOG.debug("Creating new writer factory for table '{}'",
table.name());
+ return new RowDataTaskWriterFactory(
+ () -> table,
+ FlinkSchemaUtil.convert(element.schema()),
+ flinkWriteConf.targetDataFileSize(),
+ flinkWriteConf.dataFileFormat(),
+ tableWriteProperties,
+ Lists.newArrayList(equalityFieldIds),
+ element.upsertMode(),
+ element.schema(),
+ element.spec(),
+ (partition, keyType) -> insertedRowTracker(trackerScope,
partition, keyType));
+ });
+
+ taskWriterFactory.initialize(subTaskId, attemptId);
+ return taskWriterFactory.create();
+ }
+
+ private InsertedRowTracker insertedRowTracker(
+ WriteTarget trackerScope, StructLike partition, Types.StructType
keyType) {
+ TableTrackers trackers =
+ insertedRowTrackers.computeIfAbsent(trackerScope, scope -> new
TableTrackers(keyType));
+ if (!trackers.acceptsKeyType(keyType)) {
+ LOG.warn(
+ "Not sharing inserted-row tracker for table {} branch {}: key type
{} of the earlier "
+ + "writers does not accept key type {} of the new writer (an
equality field's type, "
+ + "optionality or position changed). Re-writes of a key across
these schema versions "
+ + "within one checkpoint will produce equality deletes that
cannot apply to data "
+ + "written in the same commit",
+ trackerScope.tableName(),
+ trackerScope.branch(),
+ trackers.keyType(),
+ keyType);
+ return null;
+ }
+
+ return trackers.forPartition(partition);
+ }
+
+ /** Inserted-row trackers of one table within one checkpoint, one per
partition. */
+ private static class TableTrackers {
+ private final Types.StructType keyType;
+ private final Map<StructLike, InsertedRowTracker> byPartition =
Maps.newHashMap();
+
+ private TableTrackers(Types.StructType keyType) {
+ this.keyType = keyType;
+ }
+
+ private Types.StructType keyType() {
+ return keyType;
+ }
+
+ private boolean acceptsKeyType(Types.StructType otherKeyType) {
+ return byPartition.isEmpty()
+ ||
byPartition.values().iterator().next().acceptsKeyType(otherKeyType);
+ }
+
+ private InsertedRowTracker forPartition(StructLike partition) {
+ return byPartition.computeIfAbsent(partition, part ->
InsertedRowTracker.create(keyType));
Review Comment:
Partition column type promotion, like int->long, causes the same edge case
this PR is fixing.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/WriteTarget.java:
##########
@@ -75,6 +75,15 @@ Set<Integer> equalityFields() {
return equalityFields;
}
+ /**
+ * Returns the target with its schema ID removed. Writers whose targets
differ only in schema ID
+ * write the same table, branch and spec with the same equality fields, so
rows they write within
+ * one checkpoint can be retired by each other's position deletes.
+ */
+ WriteTarget withoutSchemaId() {
+ return new WriteTarget(tableName, branch, null, specId, upsertMode,
equalityFields);
Review Comment:
A few edge cases here:
1. Equality deletes should not be segregated by partition spec. A record
written to spec v0 can be overwritten by a record written to spec v1. Could be
caused by a partition evolution, or simply user logics routing records to
different specs depending on some criteria.
2. Same with upsertMode. An early record with upsertMode=false should still
be overwritten by a later record with upsertMode=true.
--
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]
