refactor(variant): self-align log-block variant rows, drop buffer-level projection hook - #18923
Conversation
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.
Thanks for the contribution! This PR pushes variant projection down into the log readers (parquet native projection and avro deserialization hook) and removes the buffer-level projection step, which cleanly separates Spark-4.1 variant concerns from format-agnostic merge code. One worth-checking edge case in the inline comment around how the new ordering interacts with the schema-evolution transformer when both are active. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of small readability nits below, otherwise the refactor is clean.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.
Thanks for working on this! This PR cleanly pushes the Spark 4.1 variant projection out of the engine-neutral merge buffer and into the engine-specific reader/log-block paths, which is a nice architectural improvement. No new issues flagged from this automated pass — the ordering-flip concern (schema-evolve then project → project then schema-evolve, which only matters when schema-on-read is active) was already raised and acknowledged in round 1. A Hudi committer or PMC member can take it from here for a final review. A couple of minor naming and readability suggestions below.
cc @yihua
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.
Thanks for addressing the feedback! These follow-ups rename the parquet reader's variant overload parameter to projectedStructSchema (matching Javadoc and call sites) and document why the merger != null guard in the Spark reader context is real rather than defensive. Both of my prior nits and the corresponding hudi-agent suggestions are resolved, and no new issues surfaced from this incremental pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.
Thanks for the iteration! This round's changes (renaming structSchema → projectedStructSchema, extracting isPayloadBased, and the null-guard rationale) all look good. The variant-projection refactor cleanly separates engine-specific concerns from the buffer code — Avro log blocks project in deserializeRecords via projectLogBlockRecords, parquet log blocks project natively in the reader, and the buffers go back to the plain evolved-schema transformer. No new issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review. One minor naming suggestion; the rest of the changes are clean and well-commented.
cc @yihua
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR pushes Spark 4.1 variant projection out of the engine-neutral merge buffer and into the log readers: avro log blocks rewrite rows via the new projectLogBlockRecords hook in HoodieAvroDataBlock.deserializeRecords, parquet log blocks project natively through the new getUnsafeRowIterator overload, and a single shouldProjectVariants gate drives both. I traced both read paths, both record buffers, the payload-based gate equivalence, the schema flow into the variant projector, and record count and position alignment, and the refactor looks internally consistent. The previously flagged evolve-then-project ordering change under schema-on-read remains the one item the author has already acknowledged and tested. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here., and a Hudi committer or PMC member can take it from here for a final review. A couple of minor readability suggestions below — one on an ambiguous Javadoc opener and one on comment placement.
cc @yihua
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR pushes Spark 4.1 variant projection down into the log readers so the engine-neutral FileGroupRecordBuffer no longer knows about variants: parquet log blocks project natively via a new getUnsafeRowIterator overload and avro log blocks go through a new no-op projectLogBlockRecords hook. I traced the hook wiring, the removed-hook call sites, and the common-case behavior against the old code and they line up; the evolve-vs-project ordering flip under schema-on-read is already covered in the existing inline discussion. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. - a Hudi committer or PMC member can take it from here for a final review. One naming nit on the new Javadoc overload; the rest of the refactor is clean and well-commented.
cc @yihua
…er hook Remove the engine-neutral FileGroupRecordBuffer variant-projection composition (apache#18674's getLogBlockRecordProjection hook) so the merge buffer stays format-agnostic; each log reader now emits rows already aligned to the projected read schema (apache#18739). - Parquet log blocks: thread the variant-overlaid StructType into a new HoodieSparkParquetReader.getUnsafeRowIterator(HoodieSchema, StructType, filters) overload so SPARK_ROW_REQUESTED_SCHEMA carries VariantMetadata and parquet-mr decodes variants into the projected struct shape natively (mirrors the base-file path). Wired in SparkFileFormatInternalRowReaderContext.getFileRecordIterator. - Avro log blocks: new no-op HoodieReaderContext.projectLogBlockRecords hook, invoked from HoodieAvroDataBlock.deserializeRecords; Spark overrides it to apply the VariantGet rewrite (relocated from the deleted buffer hook). - Both paths gated by a single shouldProjectVariants predicate (variant projection present AND merger not PAYLOAD_BASED), preserving the buffer's custom-payload skip. - FileGroupRecordBuffer/PositionBased now call getSchemaTransformerWithEvolvedSchema directly; getProjectedTransformer and getLogBlockRecordProjection deleted. - Sub-task 4: documented why the sparkRequiredSchema overlay must stay (HoodieSchema can't carry VariantMetadata); kept Spark-side, no schema-model change. buildVariantProjector / isVariantProjectionStruct unchanged (caller moved). Addresses apache#18739.
- HoodieSparkParquetReader: rename variant-overload parameter to structSchema and drop the no-op alias. - SparkFileFormatInternalRowReaderContext: extract isPayloadBased and drop the double negation in shouldProjectVariants.
…rger guard - Rename getUnsafeRowIterator param structSchema to projectedStructSchema - Add inline comment explaining the merger != null guard
…d sparkRequiredSchema Lead the getUnsafeRowIterator(HoodieSchema, StructType, filters) javadoc with what the overload does instead of the ambiguous "Variant overload." label, and move the rationale for the parallel sparkRequiredSchema overlay onto the constructor parameter doc where a reader first meets the field.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR pushes Spark 4.1 variant projection down into the log readers — parquet log blocks project natively via a new getUnsafeRowIterator overload, and avro log blocks rewrite via the new no-op projectLogBlockRecords hook — so the engine-neutral FileGroupRecordBuffer no longer knows about variants. I traced both buffers and both log paths (record-count preservation for the position-based buffer, iterator close chaining, and the unprojected-evolved-schema kept for the merger's read-by-ordinal contract) and didn't surface a new correctness issue in this round; the previously-flagged evolve-vs-project ordering edge case under schema-on-read has already been discussed. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. One naming nit worth a look, otherwise the refactor is clean and readable.
cc @yihua
It reads getRecordMerger(), which is null until setRecordMerger() runs during reader init, so it is state-dependent rather than a pure property; the empty parens signal that to callers per Scala convention.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR pushes Spark 4.1 variant projection down into the log readers (native parquet-mr projection for parquet log blocks, a projectLogBlockRecords hook in HoodieAvroDataBlock.deserializeRecords for avro) and removes the buffer-level getLogBlockRecordProjection hook so the engine-neutral merge buffer no longer knows about variants. I traced the gate-condition change (old payloadClasses.isPresent skip vs the new PAYLOAD_BASED merger-strategy check, which are equivalent via getPayloadClasses), the single-vs-double projection across the avro and parquet paths, position-based record-to-position alignment, and the closable-iterator chains, and found no new correctness issues in this pass. The evolve-then-project vs project-then-evolve ordering flip is already being discussed on FileGroupRecordBuffer.java, so I have left that thread to run. No new issues flagged from this automated pass, a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
…ger in the merger guard
…9 MOR table withRecordType()'s AVRO leg sets hoodie.logfile.data.block.format=avro, but on the current table version HoodieNativeLogAppendHandle takes the log format from the base file format and never consults it, so every MOR case in TestVariantDataType wrote native parquet logs and projectLogBlockRecords never ran with its gate open. Pin a MOR table to hoodie.write.table.version=9, read the log blocks back to assert they are AVRO_DATA_BLOCK (a silent fallback to parquet now fails instead of passing), and read the base + avro-log merge and a delete through cast(v as string).
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this. This PR pushes Spark 4.1 variant projection down into the log readers: parquet log blocks project natively via the new getUnsafeRowIterator(requiredSchema, structType, filters) overload, avro log blocks via the new projectLogBlockRecords hook, and the buffer-level getLogBlockRecordProjection and getProjectedTransformer are removed so FileGroupRecordBuffer no longer knows about variants. I traced the routing (parquet inline blocks reach getFileRecordIterator, avro blocks reach HoodieAvroDataBlock.deserializeRecords, default pass-through elsewhere), the payload-based gating, and the position-based buffer record-count and position invariants; they hold up, and the single-arg overload delegation is behavior-preserving. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here.; a Hudi committer or PMC member can take it from here for a final review.
. One readability nit in the test file — the listLogBlockTypes method is inserted between the listDataParquetFiles Javadoc and its method body, orphaning the comment.
cc @yihua
…ach javadoc sits on its own method
…parkSqlTestBase companion It is defined on the companion object, not the base class, so the helper added in e50c51b did not compile (scala-test-compile on hudi-spark_2.12).
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #18923 +/- ##
============================================
+ Coverage 77.95% 77.98% +0.02%
- Complexity 33447 33471 +24
============================================
Files 2539 2539
Lines 140891 140984 +93
Branches 17008 17054 +46
============================================
+ Hits 109829 109940 +111
+ Misses 23401 23382 -19
- Partials 7661 7662 +1
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
5 participants
Describe the issue this Pull Request addresses
Addresses #18739.
PR #18674 aligned Spark 4.1+ MOR variant reads via
HoodieReaderContext.getLogBlockRecordProjection, a per-row projector run inside the engine-neutralFileGroupRecordBuffer. That leaked SparkPushVariantIntoScanconcerns into format-agnostic merge code. This PR pushes variant projection down into the log readers so the buffer no longer knows about variants.Summary and Changelog
Each log reader now emits rows already aligned to the projected read schema; the buffer-level projection hook is removed.
FileGroupRecordBuffer.getProjectedTransformerandHoodieReaderContext.getLogBlockRecordProjection. The buffers (FileGroupRecordBuffer,PositionBasedFileGroupRecordBuffer) now callgetSchemaTransformerWithEvolvedSchemadirectly and keep returning the unprojected evolved schema (merger reads metadata cols by ordinal).HoodieReaderContext.projectLogBlockRecords(iter, dataBlockSchema).HoodieAvroDataBlock.deserializeRecordsinvokes it; the Spark reader context overrides it to apply theVariantGetrewrite (relocated from the old hook).HoodieSparkParquetReader.getUnsafeRowIterator(HoodieSchema, StructType, filters)overload threads the projected struct soSPARK_ROW_REQUESTED_SCHEMAcarriesVariantMetadataand parquet-mr projects natively (mirrors the base-file path).shouldProjectVariantsgate (variant projection present AND merger notPAYLOAD_BASED) drives both paths, preserving the previous custom-payload skip.No code copied.
buildVariantProjector/isVariantProjectionStructare unchanged (caller moved).Spark version scope: the projection engages on every Spark version whose adapter reports a
PushVariantIntoScanprojection struct, which today is Spark 4.1 and 4.2 (Spark4_1AdapterandSpark4_2AdapterimplementisVariantProjectionStruct/buildVariantProjectoridentically). Spark 4.0 and 3.x adapters return false and take the unprojected path. No per-version code in this PR.Rebase notes (2026-08-25, over #18961): the parquet-log leg now sits in the PARQUET arm of the log-file format dispatch from #19118/#19283 (Lance native logs keep going through
baseFileReader.readwith the overlaid schema) and uses the position-merge filter guard from #19583; the other five files merged clean.Impact
Internal change to the
HoodieReaderContextextension point (getLogBlockRecordProjection->projectLogBlockRecords); the new hook is a no-op by default. No new configs. No user-facing behavior change for non-Spark engines or Spark < 4.1; Spark 4.1 and 4.2 variant MOR reads are unchanged in result, only the alignment moves out of the buffer.Risk Level
Medium. Touches the engine-neutral merge buffer and the reader-context API. Mitigations: the new hook is a no-op default (non-Spark engines and Spark < 4.1 unaffected); projection is gated by
shouldProjectVariants(custom-payload tables skip it); the buffer still returns the unprojected evolved schema, so the merger's ordinal-based metadata access is unchanged. Verified withTestVariantDataTypeon Spark 4.0, 4.1 and 4.2 (cow + mor: insert / update / delete / merge-into,cast(v as string)selects) across key-based, position-based, and unmerged paths, plus a custom-payload variant MOR case. The 4.2 run is on the rebased head (975b687, Spark 4.2.0): 19 passed, 0 failed, the 3 Spark 3.x-only cases canceled by their version guard; it includes the log-only MOR query, MOR compaction with auto-inferred shredding, CDC and clustering cases added after #18961. CI runs the same suite on Spark 4.2 through thetest-spark-java17-scala-dml-testslane (the only 4.x version in the CI matrix). The dedicated shredded-native-log MOR legs inTestVariantShreddingMixedLayoutsstill run once #19687 lands.Documentation Update
none
Contributor's checklist