Repository navigation
Reduce string conversions and allocations in RowJsonDeserializer and RowJsonSerializer - #40770
sjvanrossum wants to merge 9 commits into
Conversation
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).
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.
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
📋 Review Summary
This PR optimizes JSON-to-Row and Row-to-JSON conversions by replacing JsonNode tree materialization in RowJsonDeserializer with streaming JsonParser token extraction, avoiding intermediate String/byte[] conversions in JsonUtils and JsonPayloadSerializerProvider, and making ObjectMapper lazy/transient in JsonUtils. It also fixes two existing bugs: custom error field schemas in JsonToRowWithErrFn and non-SQL LogicalType base/input conversions in RowJson.
🔍 General Feedback
- The streaming
JsonParserrefactoring inRowJsonandRowJsonValueExtractorsis clean, preserves existing bounds-checking and error-reporting semantics, and eliminates significant object allocation overhead. - One minor subtlety to watch out for is chained logical types (a
LogicalTypewhose base type is anotherLogicalType), which is handled inRow.getBaseValue/Row.Equalsvia awhileloop, whereasRowJsonSerializercurrently only unwraps a single level.
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
| gen.writeString(((LocalDateTime) value).toString()); // ISO 8601 format | ||
| } else { | ||
| writeValue(gen, type.getLogicalType().getBaseType(), value); | ||
| writeValue(gen, logicalType.getBaseType(), logicalType.toBaseType(value)); |
There was a problem hiding this comment.
🟢 Low: While RowJsonDeserializer recursively unwraps chained logical types via extractJsonValue(fieldName, logicalType.getBaseType(), jsonParser) (unless the inner type is one of SqlTypes.DATE/TIME/DATETIME, which expects the input type rather than the base type), RowJsonSerializer calls writeValue(gen, logicalType.getBaseType(), logicalType.toBaseType(value)). If a custom LogicalType wraps SqlTypes.DATE, SqlTypes.TIME, or SqlTypes.DATETIME as its base type, deserializer will extract a LocalDate/LocalTime/LocalDateTime and pass it to the outer logicalType.toInputType(...), which works as expected—just something to keep in mind.
64fe061 to
b6d9946
Compare
Verifies that JsonToRow.withExtendedErrorInfo() propagates custom lineField and errorField names to both the dead-letter PCollection schema and the emitted error Rows.
b6d9946 to
4f9b743
Compare
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
`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.
4f9b743 to
e44628b
Compare
|
Run Java Examples_Dataflow PreCommit |
There were previously no microbenchmarks measuring
Rowto/from JSON serialization and deserialization throughput across flat and nested/composite schemas. AddsRowJsonBenchmarkcovering bothStringandbyte[]pathways (RowJsonandJsonUtils).JsonUtilsandJsonPayloadSerializerProviderpreviously decodedbyte[]payloads into anew String(..., UTF_8)before parsing and serializedRows to aStringbefore calling.getBytes(UTF_8). Adds directbyte[]helpers onRowJsonUtils(jsonBytesToRow/rowToJsonBytes) using Jackson'sObjectReader.readValue(byte[])andObjectWriter.writeValueAsBytes(Object)to read and write UTF-8 bytes directly.RowJsonDeserializerandRowJsonSerializerallocated anAutoValueFieldValuewrapper on every field, usedStreampipelines (IntStream,StreamSupport) to iterate over arrays/rows, and usedSequentialReadsequential lists for array values. ReplacesStreampipelines andFieldValuewrappers with direct loops,ArrayListsizing, andRow.withSchema(schema).attachValues(...).RowJsonDeserializerpreviously calledjsonParser.readValueAsTree()to materialize an entire JacksonObjectNode/ArrayNodeAST in memory before extractingRowfields. Streams tokens directly fromJsonParserin a single pass withParserExtractors, caches the boundObjectReader/ObjectWriterinJsonUtils, and avoids intermediate tree allocations altogether.deserializeComplexRowBytesdeserializeComplexRowStringdeserializeFlatRowStringserializeComplexRowBytesserializeComplexRowStringNote
This PR stacks on top of #40769