mxm commented on code in PR #18675:
URL: https://github.com/apache/iceberg/pull/18675#discussion_r4539685784
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java:
##########
@@ -1184,6 +1184,61 @@ void testUpsertV3(boolean immediateUpdate) throws
Exception {
}
}
+ @ParameterizedTest
+ @ValueSource(ints = {2, 3})
+ void upsertAfterSchemaEvolution(int formatVersion) throws Exception {
+ // A restart replays the records against the already created table, which
hides failures
+ Configuration configuration = new Configuration();
+ configuration.set(RestartStrategyOptions.RESTART_STRATEGY, "none");
+ env.configure(configuration);
+
+ Map<String, String> tableProps =
+ ImmutableMap.of(TableProperties.FORMAT_VERSION,
String.valueOf(formatVersion));
+ DynamicIcebergSink.forInput(env.fromData(1))
+ // Both versions of the key come from one input element, so no
checkpoint barrier
+ // separates them
+ .generator(
+ (Integer id, Collector<DynamicRecord> out) -> {
+ out.collect(
+ upsertById(
+ SimpleDataUtil.SCHEMA, GenericRowData.of(id,
StringData.fromString("a"))));
Review Comment:
Could we upsert the key twice under the old schema?
```suggestion
out.collect(
upsertById(
SimpleDataUtil.SCHEMA, GenericRowData.of(id,
StringData.fromString("a"))));
out.collect(
upsertById(
SimpleDataUtil.SCHEMA, GenericRowData.of(id,
StringData.fromString("a2"))));
```
##########
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))
Review Comment:
NIT: Could we inline `createWriter` back into the lambda? It's left over
from the get/put version, and without it the diff here is only the
`trackerScope` line and the new argument.
--
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]
