This is an automated email from the ASF dual-hosted git repository.
stankiewicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new cbe82e50bc3 Tag the TFRecord write error output with the schema it
actually emits (#39759)
cbe82e50bc3 is described below
commit cbe82e50bc3a6d6bfc2b434db90eb36a942c891c
Author: ZIHAN DAI <[email protected]>
AuthorDate: Mon Aug 31 19:20:41 2026 +1000
Tag the TFRecord write error output with the schema it actually emits
(#39759)
ErrorFn is constructed with errorSchema = ErrorHandling.errorSchema(
inputSchema) and emits ErrorHandling.errorRecord(errorSchema, ..), so
every row on ERROR_TAG carries that schema. The collection was then
tagged with ErrorHandling.errorSchema(errorSchema) -- the wrapper
applied twice -- declaring {failed_row: {failed_row: .., error_message},
error_message} which no element it produces can match.
The sibling in the same package gets it right five lines after the same
ErrorFn construction: TFRecordReadSchemaTransformProvider:140 does
setRowSchema(errorSchema). JavaFilter, JavaMapToFields, PubsubRowToMessage
and BigQueryStorageWriteApi all do the same.
One test, and it needs no runner -- the schema is fixed when the graph is
built. Restoring the second wrap fails it and nothing else: 8 tests, 1
failure.
KafkaWriteSchemaTransformProvider:301 has the identical double wrap. It
is a different module, so it is a separate change rather than folded in
here.
---
.../io/TFRecordWriteSchemaTransformProvider.java | 3 +--
.../io/TFRecordSchemaTransformProviderTest.java | 29 ++++++++++++++++++++++
2 files changed, 30 insertions(+), 2 deletions(-)
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TFRecordWriteSchemaTransformProvider.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TFRecordWriteSchemaTransformProvider.java
index bc9b7bbeac6..c787997b1e7 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TFRecordWriteSchemaTransformProvider.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TFRecordWriteSchemaTransformProvider.java
@@ -184,8 +184,7 @@ public class TFRecordWriteSchemaTransformProvider
output = "";
}
}
- PCollection<Row> errorOutput =
-
byteArrays.get(ERROR_TAG).setRowSchema(ErrorHandling.errorSchema(errorSchema));
+ PCollection<Row> errorOutput =
byteArrays.get(ERROR_TAG).setRowSchema(errorSchema);
return PCollectionRowTuple.of(handleErrors ? output : "errors",
errorOutput);
}
}
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TFRecordSchemaTransformProviderTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TFRecordSchemaTransformProviderTest.java
index 9c067a533e0..65e38ed9d29 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TFRecordSchemaTransformProviderTest.java
+++
b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TFRecordSchemaTransformProviderTest.java
@@ -49,6 +49,7 @@ import
org.apache.beam.sdk.io.TFRecordWriteSchemaTransformProvider.TFRecordWrite
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
import org.apache.beam.sdk.schemas.transforms.SchemaTransformProvider;
+import org.apache.beam.sdk.schemas.transforms.providers.ErrorHandling;
import org.apache.beam.sdk.testing.NeedsRunner;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
@@ -238,6 +239,34 @@ public class TFRecordSchemaTransformProviderTest {
.build());
}
+ @Test
+ public void testWriteErrorSchemaMatchesTheRowsErrorFnEmits() throws
Exception {
+ // The schema is fixed when the graph is built, so this needs no runner.
+ writePipeline.enableAbandonedNodeEnforcement(false);
+
+ Schema schema = Schema.of(Schema.Field.of("record",
Schema.FieldType.BYTES));
+
+ TFRecordWriteSchemaTransformProvider provider = new
TFRecordWriteSchemaTransformProvider();
+ TFRecordWriteSchemaTransform transform =
+ (TFRecordWriteSchemaTransform)
+ provider.from(
+ TFRecordWriteSchemaTransformConfiguration.builder()
+
.setOutputPrefix(tempFolder.getRoot().toPath().resolve("errors").toString())
+ .setCompression("UNCOMPRESSED")
+ .setNumShards(0)
+ .setNoSpilling(true)
+ .build());
+
+ Row row =
Row.withSchema(schema).addValue("foo".getBytes(StandardCharsets.UTF_8)).build();
+ PCollection<Row> input =
+
writePipeline.apply(Create.of(Collections.singletonList(row)).withRowSchema(schema));
+ PCollectionRowTuple result = PCollectionRowTuple.of("input",
input).apply(transform);
+
+ // ErrorFn emits ErrorHandling.errorRecord(errorSchema, ..), so the
collection has to carry
+ // that schema. Wrapping it a second time declares a shape no element it
emits can match.
+ assertEquals(ErrorHandling.errorSchema(schema),
result.get("errors").getSchema());
+ }
+
@Test
public void testReadFindTransformAndMakeItWork() {
ServiceLoader<SchemaTransformProvider> serviceLoader =
