JTidbury2 commented on code in PR #18675:
URL: https://github.com/apache/iceberg/pull/18675#discussion_r4455997290
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriter.java:
##########
@@ -81,83 +90,157 @@ class DynamicWriter implements
CommittingSinkWriter<DynamicRecordInternal, Dynam
this.attemptId = attemptId;
this.taskWriterFactories = new LRUCache<>(cacheMaximumSize);
this.writers = Maps.newHashMap();
+ this.insertedRowTrackers = Maps.newHashMap();
LOG.debug("DynamicIcebergSinkWriter created for subtask {} attemptId {}",
subTaskId, attemptId);
}
@Override
public void write(DynamicRecordInternal element, Context context)
throws IOException, InterruptedException {
- writers
- .computeIfAbsent(
- new WriteTarget(
- element.tableName(),
- element.branch(),
- element.schema().schemaId(),
- 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();
- })
- .write(element.rowData());
+ WriteTarget writeTarget =
+ new WriteTarget(
+ element.tableName(),
+ element.branch(),
+ element.schema().schemaId(),
+ element.spec().specId(),
+ element.upsertMode(),
+ element.equalityFields());
+
+ TaskWriter<RowData> writer = writers.get(writeTarget);
+ if (writer == null) {
+ warnIfSplitsTable(writeTarget);
Review Comment:
Agreed, removed in d2fcb9999 and back to computeIfAbsent. It fired for every
second writer, append-only tables included, and only described correct
behaviour, so there was nothing to act on. I also dropped
WriteTarget.sameScopeAs, which only that warning used.
--
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]
