fix(spark): read shredded variants through the CDC and legacy streami… - #19583
fix(spark): read shredded variants through the CDC and legacy streami…#19583voonhous wants to merge 1 commit into
Conversation
…ng paths Writing the coverage apache#19578 asked for surfaced three defects of the apache#19556 null-read family on the query paths that do not go through catalyst: - HoodieMergeOnReadRDDV2 base-only splits bypass the internal reader context and read through the plain skip-merging base reader, so a shredded base file with no logs read null variants on the legacy (hoodie.file.group.reader.enabled=false) streaming/relation path. Splits with variant columns now take the file-group reader branch instead. - CDCFileGroupIterator's BASE_FILE_INSERT case reads the new base file directly with the plain table schema, so CDC after-images of insert commits read null variants under every supplemental logging mode. The direct read now applies the same full-variant rewrite and restore as the reader context (helpers factored out of SparkFileFormatInternalRowReaderContext; context behavior unchanged). - InternalRowToJsonStringConverter had no VariantType case, so CDC images serialized the VariantVal bean (raw bytes as base64) instead of the variant's JSON. Variant columns now embed as real JSON nodes via VariantVal.toString. Tests: CDC round trip over a shredded COW table (insert + update, OP_KEY_ONLY and DATA_BEFORE_AFTER, shredded layout pinned; single-row table so the in-flight avro merge-path fix apache#19582 is not a prerequisite), and a legacy-path MOR streaming round trip covering both the base-only and merged split branches over a compacted shredded base file.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR fixes shredded-variant null reads on three Spark paths that bypass the internal reader context — the legacy MOR base-only split, CDC's BASE_FILE_INSERT direct base-file read, and the CDC before/after JSON image converter — by factoring the full-variant rewrite/restore into shared helpers and applying them consistently. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
|
@wombatu-kun Can you please help to review this PR too? Thank you. |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19583 +/- ##
============================================
- Coverage 77.49% 75.78% -1.71%
+ Complexity 32799 32093 -706
============================================
Files 2522 2522
Lines 139179 139193 +14
Branches 16734 16736 +2
============================================
- Hits 107855 105488 -2367
- Misses 23748 26074 +2326
- Partials 7576 7631 +55
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
| // #19556 defect family). Splits with variant columns take the file-group reader below, whose | ||
| // reader context requests the full-variant projection shape instead (#19578). | ||
| private val requiredSchemaHasVariant: Boolean = | ||
| requiredSchema.structTypeSchema.fields.exists(f => sparkAdapter.isVariantType(f.dataType)) |
There was a problem hiding this comment.
isVariantType is also true on Spark 4.0, where buildFullVariantReadSchema returns None, so a base-only split there loses the skip-merging reader for a file-group read that does not apply the rewrite either. Gate on sparkAdapter.buildFullVariantReadSchema(requiredSchema.structTypeSchema).isDefined so the routing tracks the rewrite that motivates it.
|
|
||
| val iter: Iterator[InternalRow] = partition.split match { | ||
| case dataFileOnlySplit if dataFileOnlySplit.logFiles.isEmpty => | ||
| case dataFileOnlySplit if dataFileOnlySplit.logFiles.isEmpty && !requiredSchemaHasVariant => |
There was a problem hiding this comment.
requiredSchemaReaderSkipMerging appends partition values from the partition path when shouldExtractPartitionValuesFromPartitionPath holds, and the file-group-reader branch has no equivalent, so re-routed base-only splits would read partition columns as null. The merged branch already has that gap - is widening it to base-only splits intended here?
| structMap.toMap | ||
| case _ => value // fallback | ||
| } | ||
| case dt if SparkAdapterSupport.sparkAdapter.isVariantType(dt) => |
There was a problem hiding this comment.
This guard resolves SparkAdapterSupport.sparkAdapter for every field that is not string, array, map or struct, and hudi-spark-common has no version adapter on its test classpath, so all 11 TestInternalRowToJsonStringConverter cases now error with ClassNotFoundException: org.apache.spark.sql.adapter.Spark4_2Adapter. Detecting the variant type without the adapter (its typeName is variant) both fixes that and lets the suite cover the new case.
| // VariantVal.toString renders the variant as JSON; embed it as a real JSON node so | ||
| // the image carries the variant's structure. Falling through to the default would | ||
| // serialize the VariantVal bean, i.e. its raw value/metadata bytes as base64. | ||
| mapper.readTree(value.toString) |
There was a problem hiding this comment.
Variant.toJson renders a non-finite double as a bare NaN or Infinity token and this ObjectMapper rejects those, so a variant column holding one turns a CDC query that previously returned a base64 image into a hard failure. Worth falling back to the raw string when readTree throws - follow-up, not a blocker.
| // INMEMORY index routes MOR inserts to log files, so the first base file is the | ||
| // compaction's SHREDDED one; compact = true trips inline compaction on that write. | ||
| def addVariantData(valuesSql: String, compact: Boolean): Unit = { | ||
| spark.sql(valuesSql).write.format("org.apache.hudi") |
There was a problem hiding this comment.
This write fails in CI with UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE: the v1 DefaultSource rejects the v VARIANT column, so the test dies before the streaming read. Create and populate the table through SQL the way TestVariantDataType does.
| .load(tablePath) | ||
| .selectExpr("id", "cast(v as string) as v", "ts") | ||
|
|
||
| testStream(df)( |
There was a problem hiding this comment.
testLegacyIncrementalStreamSource asserts the executed plan is a Scan ExistingRDD with no FileScan, precisely so a silent fallback to the file-group reader is caught, and this test has no such guard. Add the same plan assertion so it cannot go green while covering the already-fixed path.
| StopStream, | ||
|
|
||
| // Merged split: the update lands in a log file on the compacted slice, so the next | ||
| // batch merges the shredded base with the log. |
There was a problem hiding this comment.
MergeOnReadIncrementalRelationV2 builds its file-system view from affectedFilesInCommits alone, so the second batch sees only the appended log file and gets a log-only split rather than base plus log. The merged branch is not covered here - worth reflecting that in the comment, or covering it another way.
| | preCombineField = 'ts', | ||
| | 'hoodie.table.cdc.enabled' = 'true', | ||
| | 'hoodie.table.cdc.supplemental.logging.mode' = '$loggingMode', | ||
| | hoodie.parquet.variant.write.shredding.enabled = 'true', |
There was a problem hiding this comment.
fullVariantReadSchemaWithOrdinals asks the adapter and never the file, so every unshredded CDC read on Spark 4.1+ now also takes the rewritten shape plus a restore projection, and no test covers that. Add an unshredded twin here, the way the clustering and bulk_insert tests already have one.
| s"Base file should carry typed_value. Schema:\n$variantGroup") | ||
| } | ||
|
|
||
| spark.sql(s"""update $tableName set v = parse_json('{"key":"value2"}'), ts = 1001 where id = 1""") |
5 participants
…ng paths
Describe the issue this Pull Request addresses
Closes #19578. That issue asked for streaming and CDC round-trip coverage of shredded variant reads; writing the coverage surfaced three real defects on those paths, all of the #19556 null-read family (requesting native VariantType against a shredded parquet base file clips the shredded group to {metadata, value} and reads value=null):
HoodieMergeOnReadRDDV2base-only splits (dataFileOnlySplit) bypass the internal reader context entirely and read through the relation's plain skip-merging base reader, so a shredded base file with no log files read null variants. Hit by the legacy (hoodie.file.group.reader.enabled=false) streaming/relation path, e.g. any streaming batch over a freshly compacted slice.CDCFileGroupIterator's BASE_FILE_INSERT case reads the new base file directly with the plain table schema, bypassing the context, so CDC after-images of insert commits read null variants. Hit under every supplemental logging mode.InternalRowToJsonStringConverter(CDC before/after images) had no VariantType handling, so a variant column fell through to Jackson bean serialization of the raw VariantVal bytes instead of the variant's JSON.Summary and Changelog
HoodieMergeOnReadRDDV2: splits whose required schema has variant columns skip the plain fast reader and take the file-group reader branch, whose reader context requests the full-variant projection shape (fix(spark): read shredded variants through internal write-side parquet reads #19558). Base-only splits without variants are unchanged.SparkFileFormatInternalRowReaderContext: the full-variant rewrite and restore projection are factored into reusable helpers (fullVariantReadSchemaWithOrdinals,variantRestoreProjection); behavior of the context itself is unchanged.CDCFileGroupIterator: BASE_FILE_INSERT applies the same rewrite/restore around its direct base-file read (with a defensive copy, since the restore projection reuses its output buffer).InternalRowToJsonStringConverter: variant columns are embedded into before/after images as real JSON nodes viaVariantVal.toString(which renders the variant as JSON), soget_json_object(after, '$.v.key')works as expected.Impact
Wrong-results fixes on non-default or CDC paths, plus one image-format improvement. CDC after-images of insert commits and legacy-path reads of base-only shredded slices returned null variants before this change. CDC images of variant columns now carry the variant's JSON structure; previously they carried a Jackson bean rendering of the raw bytes, which no reader could reasonably consume, so this is considered a fix rather than a breaking format change.
Risk Level
low. The RDD routing change only triggers when the schema has variant columns; the CDC direct-read rewrite only rewrites when the adapter supports shredded reads (Spark 4.1+), and the image converter change only affects variant columns.
Documentation Update
none
Contributor's checklist