kumarpritam863 commented on code in PR #18739:
URL: https://github.com/apache/iceberg/pull/18739#discussion_r4520102610
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicTableUpdateOperator.java:
##########
@@ -80,10 +94,35 @@ public void open(OpenContext openContext) throws Exception {
catalog,
caseSensitive,
dropUnusedColumns);
+
+ if (errorFilter != null) {
+ errorStream =
+ new OutputTag<>(
+ DynamicRecordProcessor.ERROR_STREAM,
+ new TupleTypeInfo<>(
+ new DynamicRecordInternalType(
+ catalogLoader, true, cacheMaximumSize,
inputSchemasPerTableCacheMaximumSize),
+ TypeInformation.of(Exception.class)));
+ }
}
@Override
- public DynamicRecordInternal map(DynamicRecordInternal data) throws
Exception {
+ public void processElement(
+ DynamicRecordInternal record, Context ctx,
Collector<DynamicRecordInternal> out)
+ throws Exception {
+ try {
+ out.collect(processRecord(record));
+ } catch (RuntimeException e) {
+ // Divert only failures the filter whitelists; otherwise propagate so
Flink can retry.
+ if (errorFilter != null && errorFilter.shouldDivert(e)) {
+ ctx.output(errorStream, Tuple2.of(record, e));
Review Comment:
Yes, and it was more than cosmetic. For a float input column stored as
double, the diverted record carried the table schema over the float-encoded
row, so 1.5 read back as 5.28E-315. The operator now validates and converts
first and mutates the record last, so the diverted record is the one that was
received. Test:
TestDynamicTableUpdateOperator#divertedRecordIsTheUnmodifiedInput.
--
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]
