This is an automated email from the ASF dual-hosted git repository.
damccorm 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 918363290d6 Fix correctness issue in JsonToRow, RowJsonDeserializer
and RowJsonSerializer (#40769)
918363290d6 is described below
commit 918363290d6647a72988c0118b956e0ab44e1d58
Author: Steven van Rossum <[email protected]>
AuthorDate: Mon Oct 26 19:16:45 2026 +0100
Fix correctness issue in JsonToRow, RowJsonDeserializer and
RowJsonSerializer (#40769)
* Add reproducer test for non-SQL logical types in RowJson
Adds an EnumerationType field to
RowJsonTest.ValueTests.makeLogicalTypeTestCase()
to verify serialization, deserialization, and round-tripping of logical
types
whose input type (EnumerationType.Value) differs from their base type
(INT32).
* Convert non-SQL logical types between input and base values in RowJson
RowJson only special-cased SQL date/time logical types by identifier and
fell through to the underlying base FieldType for all other logical types
without converting between the logical input type (T) and base type (BaseT),
causing ClassCastExceptions on serialize and deserialize.
* Add reproducer test for custom dead-letter field names in JsonToRow
Verifies that JsonToRow.withExtendedErrorInfo() propagates custom lineField
and errorField names to both the dead-letter PCollection schema and the
emitted error Rows.
* Use configured line and error field names in JsonToRowWithErrFn
`JsonToRowWithErrFn.expand()` hardcoded the default 'line' and 'err' field
names when registering the dead-letter PCollection schema instead of using
`lineField()` and `errorField()`, causing a schema mismatch when custom
field
names were configured.
---
.../org/apache/beam/sdk/transforms/JsonToRow.java | 43 ++++++++++++++++------
.../java/org/apache/beam/sdk/util/RowJson.java | 19 ++++++----
.../apache/beam/sdk/transforms/JsonToRowTest.java | 5 +++
.../java/org/apache/beam/sdk/util/RowJsonTest.java | 8 +++-
4 files changed, 54 insertions(+), 21 deletions(-)
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/JsonToRow.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/JsonToRow.java
index 9845af44a82..9b7ee3f4d52 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/JsonToRow.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/JsonToRow.java
@@ -261,22 +261,33 @@ public class JsonToRow {
return this.toBuilder().setNullBehavior(nullBehavior).build();
}
+ Schema getErrorSchema() {
+ if (getExtendedErrorInfo()) {
+ if (LINE_FIELD_NAME.equals(getLineFieldName())
+ && ERROR_FIELD_NAME.equals(getErrorFieldName())) {
+ return ERROR_ROW_WITH_ERR_MSG_SCHEMA;
+ }
+ return Schema.of(
+ Field.of(getLineFieldName(), FieldType.STRING),
+ Field.of(getErrorFieldName(), FieldType.STRING));
+ } else {
+ if (LINE_FIELD_NAME.equals(getLineFieldName())) {
+ return ERROR_ROW_SCHEMA;
+ }
+ return Schema.of(Field.of(getLineFieldName(), FieldType.STRING));
+ }
+ }
+
@Override
public ParseResult expand(PCollection<String> jsonStrings) {
+ Schema errorSchema = getErrorSchema();
PCollectionTuple result =
jsonStrings.apply(
- ParDo.of(ParseWithError.create(this))
+ ParDo.of(ParseWithError.create(this, errorSchema))
.withOutputTags(PARSED_LINE, TupleTagList.of(PARSE_ERROR)));
- PCollection<Row> failures;
-
- if (getExtendedErrorInfo()) {
- failures =
-
result.get(PARSE_ERROR).setRowSchema(JsonToRowWithErrFn.ERROR_ROW_WITH_ERR_MSG_SCHEMA);
- } else {
- failures =
result.get(PARSE_ERROR).setRowSchema(JsonToRowWithErrFn.ERROR_ROW_SCHEMA);
- }
+ PCollection<Row> failures =
result.get(PARSE_ERROR).setRowSchema(errorSchema);
return ParseResult.resultBuilder()
.setCallingPipeline(jsonStrings.getPipeline())
@@ -292,18 +303,28 @@ public class JsonToRow {
public abstract JsonToRowWithErrFn getJsonToRowWithErrFn();
+ public abstract Schema getErrorSchema();
+
public abstract Builder toBuilder();
@AutoValue.Builder
public abstract static class Builder {
public abstract Builder setJsonToRowWithErrFn(JsonToRowWithErrFn
value);
+ public abstract Builder setErrorSchema(Schema value);
+
public abstract ParseWithError build();
}
public static ParseWithError create(JsonToRowWithErrFn
jsonToRowWithErrFn) {
+ return create(jsonToRowWithErrFn, jsonToRowWithErrFn.getErrorSchema());
+ }
+
+ public static ParseWithError create(
+ JsonToRowWithErrFn jsonToRowWithErrFn, Schema errorSchema) {
return new
AutoValue_JsonToRow_JsonToRowWithErrFn_ParseWithError.Builder()
.setJsonToRowWithErrFn(jsonToRowWithErrFn)
+ .setErrorSchema(errorSchema)
.build();
}
@@ -317,14 +338,14 @@ public class JsonToRow {
output
.get(PARSE_ERROR)
.output(
- Row.withSchema(ERROR_ROW_WITH_ERR_MSG_SCHEMA)
+ Row.withSchema(getErrorSchema())
.addValue(element)
.addValue(ex.getMessage())
.build());
} else {
output
.get(PARSE_ERROR)
-
.output(Row.withSchema(ERROR_ROW_SCHEMA).addValue(element).build());
+
.output(Row.withSchema(getErrorSchema()).addValue(element).build());
}
return;
}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/RowJson.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/RowJson.java
index ccfbd87d450..82def4b105f 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/RowJson.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/RowJson.java
@@ -321,7 +321,9 @@ public class RowJson {
}
if (fieldValue.typeName().isLogicalType()) {
- String identifier = fieldValue.type().getLogicalType().getIdentifier();
+ Schema.LogicalType<Object, Object> logicalType =
+ (Schema.LogicalType<Object, Object>)
fieldValue.type().getLogicalType();
+ String identifier = logicalType.getIdentifier();
if (SqlTypes.DATE.getIdentifier().equals(identifier)) {
return dateValueExtractor().extractValue(fieldValue.jsonValue());
} else if (SqlTypes.TIME.getIdentifier().equals(identifier)) {
@@ -329,11 +331,10 @@ public class RowJson {
} else if (SqlTypes.DATETIME.getIdentifier().equals(identifier)) {
return
localDatetimeValueExtractor().extractValue(fieldValue.jsonValue());
} else {
- return extractJsonNodeValue(
- FieldValue.of(
- fieldValue.name(),
- fieldValue.type().getLogicalType().getBaseType(),
- fieldValue.jsonValue()));
+ return logicalType.toInputType(
+ extractJsonNodeValue(
+ FieldValue.of(
+ fieldValue.name(), logicalType.getBaseType(),
fieldValue.jsonValue())));
}
}
@@ -602,7 +603,9 @@ public class RowJson {
gen.writeEndObject();
break;
case LOGICAL_TYPE:
- String identifier = type.getLogicalType().getIdentifier();
+ Schema.LogicalType<Object, Object> logicalType =
+ (Schema.LogicalType<Object, Object>) type.getLogicalType();
+ String identifier = logicalType.getIdentifier();
if (SqlTypes.DATE.getIdentifier().equals(identifier)) {
gen.writeString(((LocalDate) value).toString()); // ISO 8601 format
} else if (SqlTypes.TIME.getIdentifier().equals(identifier)) {
@@ -610,7 +613,7 @@ public class RowJson {
} else if (SqlTypes.DATETIME.getIdentifier().equals(identifier)) {
gen.writeString(((LocalDateTime) value).toString()); // ISO 8601
format
} else {
- writeValue(gen, type.getLogicalType().getBaseType(), value);
+ writeValue(gen, logicalType.getBaseType(),
logicalType.toBaseType(value));
}
break;
default:
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/JsonToRowTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/JsonToRowTest.java
index 79918930e09..3c66937da0e 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/JsonToRowTest.java
+++
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/JsonToRowTest.java
@@ -17,6 +17,9 @@
*/
package org.apache.beam.sdk.transforms;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
import java.io.Serializable;
import java.util.Collection;
import java.util.stream.Collectors;
@@ -317,6 +320,8 @@ public class JsonToRowTest implements Serializable {
PCollection<Row> personRows = results.getResults();
PCollection<Row> errorsWithMsg = results.getFailedToParseLines();
+ assertThat(errorsWithMsg.getSchema(), equalTo(customSchema));
+
PAssert.that(personRows).containsInAnyOrder(PERSON_ROWS);
PAssert.that(errorsWithMsg)
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/RowJsonTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/RowJsonTest.java
index fda8fea5480..6476ff4d78a 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/RowJsonTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/RowJsonTest.java
@@ -34,6 +34,7 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.Schema.FieldType;
+import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType;
import org.apache.beam.sdk.schemas.logicaltypes.VariableString;
import org.apache.beam.sdk.util.RowJson.RowJsonDeserializer;
import org.apache.beam.sdk.util.RowJson.RowJsonDeserializer.NullBehavior;
@@ -134,15 +135,18 @@ public class RowJsonTest {
}
private static Object[] makeLogicalTypeTestCase() {
+ EnumerationType enumType = EnumerationType.create("ZERO", "ONE", "TWO");
Schema schema =
Schema.builder()
.addLogicalTypeField("f_passThroughString",
VariableString.of(10))
+ .addLogicalTypeField("f_enum", enumType)
.build();
// fixed string will do padding
- String rowString = "{\n" + "\"f_passThroughString\" : \"hello\"\n" + "}";
+ String rowString =
+ "{\n" + "\"f_passThroughString\" : \"hello\",\n" + "\"f_enum\" :
1\n" + "}";
- Row expectedRow = Row.withSchema(schema).addValues("hello").build();
+ Row expectedRow = Row.withSchema(schema).addValues("hello",
enumType.valueOf("ONE")).build();
return new Object[] {"Logical Types", schema, rowString, expectedRow};
}
