mxm commented on code in PR #18739:
URL: https://github.com/apache/iceberg/pull/18739#discussion_r4491389240
##########
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:
Is this a regression for upsert records without explicit equality fields?
`DynamicWriter` falls back to the table's identifier fields when the record has
none, but `data.equalityFields()` is only resolved against the record schema.
Could we apply the same fallback to `tableSchema.identifierFieldIds()`
before validating?
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java:
##########
@@ -134,6 +139,15 @@ static DataConverter nullable(DataConverter converter) {
return value -> value == null ? null : converter.convert(value);
}
+ static DataConverter nonNullable(DataConverter converter, String fieldName) {
+ return value -> {
+ if (value == null) {
+ throw new IllegalArgumentException("Null value for non-nullable field:
" + fieldName);
Review Comment:
Does this break upsert deletes? In upsert mode a `RowKind.DELETE` row only
needs the equality fields, the delta writer reads just the key for it.
Should we skip this check for DELETE rows?
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java:
##########
@@ -130,7 +149,31 @@ public void processElement(T element, Context ctx,
Collector<DynamicRecordIntern
@Override
public void collect(DynamicRecord inputData) {
DynamicRecordWithConfig data = dynamicRecordWithConfig.wrap(inputData);
+ try {
+ processRecord(data);
+ } catch (RuntimeException e) {
+ // Divert only failures the filter whitelists; otherwise propagate so
Flink can retry.
+ if (errorFilter != null && errorFilter.shouldDivert(e)) {
+ context.output(
+ errorStream,
+ Tuple2.of(
+ new DynamicRecordInternal(
+ data.tableIdentifier().toString(),
+ data.branch(),
+ data.schema(),
+ data.rowData(),
+ data.spec(),
+ -1,
+ data.upsertMode(),
+ DynamicSinkUtil.getEqualityFieldIds(data.equalityFields(),
data.schema())),
Review Comment:
`getEqualityFieldIds` throws when an equality field is not in the schema, so
building the error record can fail again.
Could we build the error record without resolving the field IDs again?
##########
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:
`processRecord` mutates `record` before `validateUpsert` and the conversion
run, so the diverted record has the table's schema and spec but still the
unconverted row.
Could we divert the unmodified record?
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java:
##########
@@ -2143,6 +2145,80 @@ void
testWriteOldShapeRecordAfterColumnReAddedWithWiderType() throws Exception {
assertThat(records.get(0).getField("c")).isEqualTo(5L);
}
+ @Test
+ void testSideOutputForErroneousRecordsImmediateUpdatePath() {
+ runTestSideOutputForErroneousRecords(true);
+ }
+
+ @Test
+ void testSideOutputForErroneousRecordsNonImmediateUpdatePath() {
+ runTestSideOutputForErroneousRecords(false);
+ }
+
+ private void runTestSideOutputForErroneousRecords(boolean immediateUpdate) {
+ // Converting int -> boolean will fail
+ Schema incompatibleSchemaChange =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.BooleanType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()));
+
+ List<DynamicIcebergDataImpl> rows =
+ Lists.newArrayList(
+ new DynamicIcebergDataImpl(
+ SimpleDataUtil.SCHEMA,
+ "t1",
+ SnapshotRef.MAIN_BRANCH,
+ PartitionSpec.unpartitioned()),
+ new DynamicIcebergDataImpl(
+ // Converting int -> boolean will fail
+ incompatibleSchemaChange,
+ "t1",
+ SnapshotRef.MAIN_BRANCH,
+ PartitionSpec.unpartitioned(),
+ true,
+ Collections.singleton("id"),
+ false));
+
+ DataStream<DynamicIcebergDataImpl> dataStream =
+ env.fromData(rows, TypeInformation.of(new TypeHint<>() {}));
+ env.setParallelism(1);
+
+ AtomicInteger errorCount = new AtomicInteger();
+ AtomicReference<Tuple2<DynamicRecord, Exception>> rejectedRecord = new
AtomicReference<>();
+
+ DynamicIcebergSink.forInput(dataStream)
+ .generator(new Generator())
+ .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+ .errorStreamConsumer(
+ errorStream -> {
+ // This code runs at the end of the append() call to execute the
Flink job and
+ // retrieve the error stream.
+ // The try/catch is only required due to the checked exception
of executeAndCollect().
+ try {
+ errorStream
+ .executeAndCollect()
+ .forEachRemaining(
+ record -> {
+ errorCount.incrementAndGet();
+ rejectedRecord.set(record);
+ });
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .errorStreamFilter(error -> true)
+ .immediateTableUpdate(immediateUpdate)
+ .append();
+
+ assertThat(errorCount.get()).isEqualTo(1);
Review Comment:
Could we also assert that the first (valid) record ends up in `t1`?
Otherwise the test doesn't show that the job keeps running after a diverted
record.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java:
##########
@@ -172,7 +172,8 @@ private static DynamicRecordProcessor<DynamicRecord>
processorWithoutImmediateUp
String.valueOf(CACHE_REFRESH_MS)),
new Configuration()),
Collections.emptyMap(),
- new Configuration());
+ new Configuration(),
+ null);
Review Comment:
Could we add processor tests here, using the harness like the other tests in
this class? Only the end-to-end test covers the new catch block today. I'd like
to see a diverted failure and a propagated failure. And one with forward write
path.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -298,6 +309,37 @@ public Builder<T> tableCreator(TableCreator
tableCreationFunction) {
return this;
}
+ /**
+ * Set a consumer for the error stream of records that fail during
conversion/table update. When
+ * set, a record that fails in the generator, schema evolution, or table
update is routed to
Review Comment:
NIT: Could we adjust "fails in the generator"? An exception thrown by
`DynamicRecordGenerator.generate` itself is not diverted.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java:
##########
@@ -130,7 +149,31 @@ public void processElement(T element, Context ctx,
Collector<DynamicRecordIntern
@Override
public void collect(DynamicRecord inputData) {
DynamicRecordWithConfig data = dynamicRecordWithConfig.wrap(inputData);
+ try {
+ processRecord(data);
Review Comment:
The try also covers the emit. This will create issue with the chained
operators for the forward write path. Ideally, we want to exclude the emission.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java:
##########
@@ -161,8 +175,14 @@ class RowDataConverter implements DataConverter {
RowType.RowField sourceField =
sourceType.getFields().get(sourceFieldIndex);
fieldGetter = RowData.createFieldGetter(sourceField.getType(),
sourceFieldIndex);
dataConverter =
- DataConverter.getNullable(
- sourceField.getType(), targetField.getType(), caseSensitive);
+ targetField.getType().isNullable()
+ ? DataConverter.getNullable(
+ sourceField.getType(), targetField.getType(),
caseSensitive)
+ : DataConverter.getNonNullable(
+ sourceField.getType(),
+ targetField.getType(),
+ caseSensitive,
+ targetField.getName());
Review Comment:
If the record schema matches the table schema, `TableMetadataCache` returns
`DataConverter.identity()` and this check never runs. E.g. table `(id required,
data required)` with a record `(1, null)` of the same schema is neither
rejected nor diverted. Should we cover that case too, or narrow the "never
reaching the writer" comments?
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordExceptionFilter.java:
##########
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink.dynamic;
+
+import java.io.Serializable;
+import org.apache.flink.annotation.Experimental;
+
+/**
+ * Classifies a failure raised while processing a {@link DynamicRecord} as
divertable or not. Keeps
+ * error handling fail-safe: only failures this filter whitelists are diverted
to the error stream
+ * set via {@link DynamicIcebergSink.Builder#errorStreamConsumer}; every other
failure propagates
+ * and fails the job so Flink's restart strategy can retry it.
+ *
+ * <p>Whitelist only deterministic, record-scoped failures (bad data, for
example a value that
+ * cannot be encoded for its table, or an upsert record whose partition source
is not an equality
+ * field). Return {@code false} for transient or systemic failures (an
unreachable catalog,
+ * object-store errors, an interrupt) so a sustained outage does not silently
drain the whole
+ * stream.
+ */
+@Experimental
+@FunctionalInterface
+public interface DynamicRecordExceptionFilter extends Serializable {
+
+ /**
+ * Classifies a failure as divertable or not.
+ *
+ * @param error the cause thrown while processing a record
+ * @return {@code true} to divert the record, {@code false} to propagate the
failure and fail the
+ * job
+ */
+ boolean shouldDivert(Exception error);
Review Comment:
How would a user write this filter? Should we add a dedicated exception type
for record-scoped failures?
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicSinkUtil.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink.dynamic;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.Collections;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+
+class TestDynamicSinkUtil {
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, "id", Types.IntegerType.get()), optional(2, "data",
Types.StringType.get()));
+
+ @Test
+ void testValidateUpsertSkippedWhenNotUpsert() {
+ assertThatCode(
+ () ->
+ DynamicSinkUtil.validateUpsert(
+ false, PartitionSpec.unpartitioned(),
Collections.emptySet()))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testValidateUpsertFailsWithoutEqualityFields() {
+ assertThatThrownBy(
+ () ->
+ DynamicSinkUtil.validateUpsert(
+ true, PartitionSpec.unpartitioned(),
Collections.emptySet()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Equality field columns shouldn't be empty when configuring to use
UPSERT data.");
Review Comment:
The old test covered more of the execution path, see
https://github.com/apache/iceberg/pull/18739/changes#diff-749232de0cb8516114173b69933365184b8ae963be6527e09096028febfb8655L283
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -589,6 +667,32 @@ private Configuration fromReadableConfig() {
? (Configuration) readableConfig
: Configuration.fromMap(readableConfig.toMap());
}
+
+ /**
+ * Rebuilds the public {@link DynamicRecord} handed to the error stream
from the internal
+ * record, preserving {@code upsertMode} and {@code equalityFields}
(mapping the internal
+ * field-ids back to column names via the record schema). Routing hints
(distribution mode,
+ * write parallelism) are not meaningful for an errored record and are
left at their defaults.
+ */
+ private static DynamicRecord toDynamicRecord(DynamicRecordInternal
internal) {
+ Schema schema = internal.schema();
+ DynamicRecord record =
+ new DynamicRecord(
+ TableIdentifier.parse(internal.tableName()),
+ internal.branch(),
+ schema,
+ internal.rowData(),
+ internal.spec(),
+ null,
+ -1);
Review Comment:
NIT: Could we use the five-argument `DynamicRecord` constructor? It sets
`null` and `-1` already.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicTableUpdateOperator.java:
##########
@@ -271,6 +285,126 @@ void testDynamicTableUpdateOperatorDropUnusedColumns()
throws Exception {
assertThat(input).isEqualTo(output);
}
+ @Test
+ void testDynamicTableUpdateOperatorErrorStream() throws Exception {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier table = TableIdentifier.of(TABLE);
+
+ AtomicInteger errorCount = new AtomicInteger();
+ AtomicReference<Tuple2<DynamicRecordInternal, Exception>> errorRecord =
new AtomicReference<>();
+
+ DynamicTableUpdateOperator operator =
+ new DynamicTableUpdateOperator(
+ CATALOG_EXTENSION.catalogLoader(),
+ TableCreator.DEFAULT,
+ flinkDynamicSinkConfiguration(CASE_SENSITIVE, PRESERVE_COLUMNS),
+ error -> true);
+ operator.open(null);
+
+ catalog.createTable(table, SCHEMA1);
+
+ // Incompatible evolution (int -> boolean) fails the update; the filter
whitelists it, so the
+ // record is routed to the side output instead of failing the job.
+ Schema incompatibleSchema =
+ new Schema(Types.NestedField.required(1, "id",
Types.BooleanType.get()));
+ DynamicRecordInternal input =
+ new DynamicRecordInternal(
+ TABLE,
+ "branch",
+ incompatibleSchema,
+ GenericRowData.of(true),
+ PartitionSpec.unpartitioned(),
+ 42,
+ false,
+ Collections.emptySet());
+
+ TestCollector collector = new TestCollector();
+ ProcessFunction<DynamicRecordInternal, DynamicRecordInternal>.Context
mockContext =
+ Mockito.mock(ProcessFunction.Context.class);
Review Comment:
Could we use `ProcessFunctionTestHarnesses.forProcessFunction` and
`harness.getSideOutput(...)`, as in `TestDynamicRecordProcessor`? I'd prefer
not to mock the `Context`, and this is the only test class in the package that
uses Mockito. A harness test could also cover a failure after the update, which
the tests here don't.
--
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]
