kumarpritam863 commented on code in PR #18739:
URL: https://github.com/apache/iceberg/pull/18739#discussion_r4520087292
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicTableUpdateOperator.java:
##########
@@ -98,6 +137,10 @@ public DynamicRecordInternal map(DynamicRecordInternal
data) throws Exception {
data.setSchema(tableSchema);
data.setSpec(newData.f1);
+ // Validate before writing: a bad upsert config or a record that cannot be
converted to the
+ // resolved schema throws here and is diverted/failed at this stage, never
reaching the writer.
+ DynamicSinkUtil.validateUpsert(data.upsertMode(), newData.f1,
data.equalityFields());
Review Comment:
Yes, this was a regression. With immediateUpdate=false, an upsert record
without equality fields for a table that has identifier fields failed with
"Equality field columns shouldn't be empty", while main writes it. The update
operator now falls back to the identifier fields of the table schema before
validating, the way DynamicWriter does. Test:
TestDynamicTableUpdateOperator#upsertFallsBackToIdentifierFieldsOfTable.
--
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]
