fix(variant): align Spark 4.1 MOR merge with PushVariantIntoScan and restore Spark 4.0 reads by voonhous · Pull Request #18674 · apache/hudi

voonhous · 2026-05-01T21:42:49Z

Describe the issue this Pull Request addresses

TestVariantDataType was previously skipped on Spark 4.1 because the MOR merge path crashed with a JVM SIGBUS during UnsafeRow.copy (translated to java.lang.InternalError: a fault occurred in a recent unsafe memory access with -XX:+UnlockDiagnosticVMOptions -XX:+PreserveFramePointer). While investigating that, two adjacent Spark 4.0 variant-read regressions and a shared-module unit-test classpath fragility surfaced; this PR fixes all three so variant + MOR works end-to-end on both Spark 4.0 and Spark 4.1.

Summary and Changelog

Re-enables TestVariantDataType on Spark 4.1 (Seq("cow", "mor")) and fixes the three latent variant-read issues uncovered while investigating #18605:

1. Spark 4.1 MOR merge — log records misaligned with PushVariantIntoScan

Spark 4.1's PushVariantIntoScan Catalyst rule rewrites a Variant column on the read schema into struct<0:..> with VariantMetadata. Base files use parquet-mr's native projection from that metadata, but log-block records reach the merger carrying the full variant, so the merger combines mismatched-shape rows. The resulting UnsafeRow has a malformed size header that survives the read pipeline and overruns memory the first time something calls UnsafeRow.copy() (in our case, RangePartitioner sampling).

HoodieReaderContext.getLogBlockRecordProjection: new engine-neutral hook (default no-op) that returns a per-row transformer aligning log-block records to the engine's projected read schema.
FileGroupRecordBuffer.getProjectedTransformer: composes schema evolution with the new hook. Returns the evolved data-block schema (not readerSchema) because the projector preserves field shape and the merger reads metadata cols (_hoodie_record_key, _tmp_metadata_row_index) by ordinal. Skipped when a custom payload class is configured (PayloadUpdateProcessor.handleNonDeletes's convertToAvroRecord would mis-decode a row whose variant has been rewritten).
PositionBasedFileGroupRecordBuffer.processDataBlock routes through the new helper.
Spark4_1Adapter.buildVariantProjector: builds a VariantGet-driven UnsafeProjection from the data-block schema to the projected required schema (the existing implementation, kept).
SparkFileFormatInternalRowReaderContext.getLogBlockRecordProjection overrides the new hook for Spark 4.1, building a target schema that preserves every field of the data-block schema and only rewrites variant fields into their PushVariantIntoScan struct shape.
SparkFileFormatInternalRowReaderContext.getFileRecordIterator builds the base-file read schema from the engine's augmented requiredSchema (so _hoodie_record_key is read for the merger's key extraction) and overlays the projected variant struct shape from sparkRequiredSchema (so parquet-mr still does the native variant projection). Drops the now-redundant lazy variantLogProjector from the log-file path; the merge buffer's hook handles it.
HoodieFileGroupReaderBasedFileFormat.supportBatch disables vectorized reads when a VariantType field is in the schema — Spark 4.1's vectorized variant path produces UnsafeRow encodings that crash during shuffle copy. Gated on gteqSpark4_1 so Spark 4.0 keeps vectorized variant reads.

2. Spark 4.0 — MALFORMED_VARIANT on legacy file format and log-block reads

#18334 moved the [value, metadata] variant field reorder out of the shared HoodieParquetReadSupport and into the version-specific Spark40HoodieParquetReadSupport, but two consumers still constructed the base class directly. With no reorder, parquet-mr hands [metadata, value] bytes to Spark 4.0's positional ParquetUnshreddedVariantConverter and Variant.<init> raises MALFORMED_VARIANT.

Spark40LegacyHoodieParquetFileFormat: switch the parquet-mr fallback branch to Spark40HoodieParquetReadSupport. Affects the legacy file-format read path on Spark 4.0.
HoodieSparkParquetReader (shared, used for parquet log-block reads on every Spark version): route through a new SparkAdapter.createParquetReadSupport factory. Default returns the base class; Spark4_0Adapter overrides to return the Spark40 subclass. Other Spark versions are unchanged.

3. Shared-module unit test classpath fragility

TestHoodieSparkSchemaUtils (in hudi-spark-common) was failing with ClassNotFoundException for Spark{3_5,4_1}Adapter after the PushVariantIntoScan support lande: every StructType conversion was forcing SparkAdapterSupport.sparkAdapter to resolve a version-specific adapter class that isn't on hudi-spark-common's test classpath.

HoodieSparkSchemaConverters.toHoodieTypeNested hoists the variant-projection guard into a isSparkVariantProjectionStruct helper that short-circuits on Spark < 4.1 and on structs whose fields carry no Spark Metadata (PushVariantIntoScan always tags every field with VariantMetadata, so empty-metadata structs are definitively not projections). Belt-and-suspenders try/catch around the adapter lookup for any other environment that hits the gap. Drops a debug LOG.warn left over from the Ensure Variant type support on Spark 4.1 #18605 investigation.

Misc

HoodieAvroUtils.createNewSchemaField: drop the diagnostic try/catch + LOG.error left over from the Ensure Variant type support on Spark 4.1 #18605 investigation, along with the now-unused slf4j Logger imports and LOG field.
HoodieFileGroupReaderBasedFileFormat: demote four logInfo debug calls (entry log + per-conversion logs + isSplitable summary) to logDebug. They fire on every reader build and would flood production logs at INFO.
Spark4_1Adapter.buildVariantProjector: replace bare sparkDataSchema.fieldIndex(...) calls with a lookupDataField helper that raises an IllegalStateException naming both schemas, so future projector mismatches are easier to triage than Spark's bare IllegalArgumentException. Drop the unused .zipWithIndex.
SparkFileFormatInternalRowReaderContext: new findFieldByName helper honoring spark.sql.caseSensitive for the variant-projection field lookups, replacing exact == matches.
AvroSchemaConverterWithTimestampNTZ: LOG.debug line when a group's LogicalTypeAnnotation is unrecognized and we fall back to record conversion. Right for the variant binary group we write; the log is so a future unrecognized annotation that isn't a record doesn't fail silently.

Impact

User-facing impact:

Variant + MOR queries (insert / update / delete / merge into / select with cast(v as string)) now work on Spark 4.1.
Spark 4.0 variant reads via the legacy file format and parquet log blocks no longer raise MALFORMED_VARIANT.
TestHoodieSparkSchemaUtils runs on shared modules without a version-specific adapter on the classpath.

Performance:

Spark 4.1 with VariantType fields is forced row-based on the file group reader path; vectorized variant reads need a separate fix to Hudi's UnsafeRow encoding (out of scope for this PR).
Spark 4.0 unchanged — the new gate (gteqSpark4_1) avoids degrading it to row-based.
Custom-payload tables with variant on Spark 4.1 skip the merge-path variant projection (degrades to the same correctness path the rest of the merge pipeline gives that combo, no crash). No effect on standard mergers.

Risk Level

Medium.

The changes touch the engine-neutral merge buffer (FileGroupRecordBuffer, PositionBasedFileGroupRecordBuffer) and the shared HoodieReaderContext API. New behavior is gated:

The new getLogBlockRecordProjection hook defaults to Option.empty(); non-Spark engines (Avro, Flink, Lance) and Spark versions other than 4.1 see no behavior change in the buffer.
The base-file read schema augmentation in SparkFileFormatInternalRowReaderContext.getFileRecordIterator is a cross-version change but only modifies what columns are read, not how the merger interprets them — the merger's key extraction was already expecting metadata cols, this just makes them actually arrive in the base record.
The Spark 4.0 ReadSupport rewiring is a subclass swap, no API change.

Verification:

TestVariantDataType passes end-to-end on Spark 4.1 (cow + mor iterations: insert, update, delete, merge into, all cast(v as string) selects).
TestVariantDataType continues to pass on Spark 4.0.
TestHoodieSparkSchemaUtils passes on Spark 3.5 and Spark 4.1 profiles.

Documentation Update

None.

Contributor's checklist

Read through contributor's guide
Enough context is provided in the sections above
Adequate tests were added if applicable

hudi-agent

🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.

Thanks for working on this! The PR fixes #18605 by routing the Spark 4.1 PushVariantIntoScan-projected struct shape through the read path, disabling vectorized reads for variant types, and moving the parquet variant-field reorder workaround to a Spark 4.0-specific class. A few items worth a closer look in the inline comments — mostly debug logging that looks like it was left in for triage and a helper that appears unused. The PR description and title also still have TODOs/CI-triggering notes that should be cleaned up. 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 few log-level and naming nits below.

hudi-agent

🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.

Thanks for working on this! The PR fixes the Spark 4.1 SIGBUS crash on variant-column reads by aligning log-block records to PushVariantIntoScan's projected struct shape and disabling vectorized reads on the variant path. Two concerns worth a closer look in the inline comments — one potential perf regression on Spark 4.0, and one latent crash in the payload-based merge path that the variant test doesn't exercise. Please take a look at the inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A few naming and diagnostic-logging suggestions below — the main concern is two verbose logging additions (LOG.warn in HoodieSparkSchemaConverters and several logInfo calls in HoodieFileGroupReaderBasedFileFormat) that look like debug aids left in from the bug investigation and will produce significant log noise in production.

hudi-agent

🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.

Thanks for the iteration! This pass moves the variant reorder from the base HoodieParquetReadSupport into a new Spark-4.0-specific subclass and routes the engine-neutral HoodieSparkParquetReader through a SparkAdapter.createParquetReadSupport SPI — Spark 4.1 inherits the no-reorder default (it reads variants by name via SPARK-54410) and Spark 4.0 keeps the workaround. The new isSparkVariantProjectionStruct helper also adds reasonable short-circuits for the gteqSpark4_1/empty-metadata cases. The earlier LOG.warn was dropped — thanks for that. The other prior findings still appear open from this pass: the bug-level concern on HoodieFileGroupReaderBasedFileFormat.scala:163 (the isVariantType branch isn't gated by gteqSpark4_1, so Spark 4.0 still loses vectorized variant reads), the question on FileGroupRecordBuffer.java:308 re: the PayloadUpdateProcessor path, plus three readability/nits (the three logInfos in buildReaderWithPartitionValues, the schemaTransformerWithEvolvedSchema variable name in PositionBasedFileGroupRecordBuffer, and the discarded zipWithIndex in Spark4_1Adapter). No new issues flagged from this incremental pass — please take a look at the still-open earlier comments, and this should be ready for a Hudi committer or PMC member to take it from here.

hudi-agent

🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.

Thanks for working on this! The PR re-enables variant tests on Spark 4.1, fixes the SIGBUS crash in the MOR merge path by aligning log-record shape with PushVariantIntoScan, and cleanly separates the Spark 4.0 variant-field reorder workaround into its own subclass. The major concerns flagged in prior rounds (debug logging, graceful fallback, case-sensitivity, payload-class early return) appear addressed. 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 couple of small readability suggestions below — overall the code is clean and well-commented.

cc @yihua

hudi-agent

🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.

Thanks for the thorough work on this! The PR fixes the Spark 4.1 MOR variant merge SIGBUS by aligning log-block records with PushVariantIntoScan's projected shape, restores Spark 4.0 variant reads via a dedicated subclass, and re-enables TestVariantDataType on Spark 4.1. After tracing through the critical paths (the projection flow in getLogBlockRecordProjection/buildVariantProjector, the schema-vs-row shape comments in getProjectedTransformer, and the Spark 4.0 reorder split into Spark40HoodieParquetReadSupport), the architecture holds together — BoundReference indices and UnsafeProjection input attributes are consistent, payload-based merges are correctly skipped, and .copy() guards against UnsafeProjection's reused buffer. 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 few small naming and documentation nits below.

cc @yihua

hudi-agent · 2026-05-02T19:24:54Z

+    }))
+    if (!needsProjection) {
+      return HOption.empty[JFunction[InternalRow, InternalRow]]()
+    }


🤖 nit: could you rename req to sparkReq? A few lines down (in getFileRecordIterator) the same unwrapped value is bound as sparkReq in the pattern match, so using a different name here makes the two call sites harder to connect at a glance.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

hudi-agent · 2026-05-02T19:24:54Z

+   */
+  private def isSparkVariantProjectionStruct(st: StructType): Boolean = {
+    if (!HoodieSparkUtils.gteqSpark4_1) return false
+    if (!st.fields.exists(_.metadata != Metadata.empty)) return false


🤖 nit: could you flip this to if (st.fields.forall(_.metadata == Metadata.empty)) return false? The double-negative !exists(!= empty) takes a moment to mentally resolve compared to forall(== empty).

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

hudi-agent · 2026-05-02T19:24:54Z

+  /**
+   * Composes schema evolution then the engine's optional log-block record projection
+   * (currently only Spark 4.1's PushVariantIntoScan). Returns the evolved data-block schema
+   * — the projector preserves field shape, only rewriting variant fields, so merger


🤖 nit: "Returns the evolved data-block schema" reads as if the return type is HoodieSchema, but the method signature is Pair<Function<T,T>, HoodieSchema>. Could you rephrase to something like "Returns a pair of (composed transformer, evolved data-block schema); the schema component is the evolved, not post-projection, shape because..."?

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

Spark 4.1's PushVariantIntoScan Catalyst rule rewrites a VariantType column on the read schema into a struct<...> whose child fields carry VariantMetadata describing pushed-down extractions. Hudi's file format previously crashed during HoodieSparkSchemaConverters.toHoodieType because the projection's child names (e.g. "0") are invalid Avro identifiers. Add detection + handling so the projected schema flows through Hudi: - SparkAdapter.isVariantProjectionStruct / buildVariantProjector (default no-ops; Spark4_1Adapter overrides via VariantMetadata.isVariantStruct and a VariantGet-driven UnsafeProjection). - HoodieSparkSchemaConverters maps a projected variant struct to a regular HoodieSchema.Variant so internal merging / schema-handler logic is unchanged. - BaseSpark4Adapter.isDataTypeEqualForPhysicalSchema treats a projected variant struct as compatible with VariantType, preventing the implicit schema-change machinery from rewriting requiredSchema back to VariantType (which would lose the projection metadata). - SparkFileFormatInternalRowReaderContext accepts the original Spark requiredSchema / dataStructType, prefers them over the HoodieSchema-derived StructType so parquet-mr sees VariantMetadata and projects natively (also shredding-ready: when shredded variants land, parquet-mr will skip non- projected columns automatically); wraps log-file iterators with the projector so MOR log records (which carry the full variant) align to the projected shape before reaching the merger. A 5-arg auxiliary constructor preserves Java callers. - HoodieFileGroupReaderBasedFileFormat.supportBatch disables vectorized reading when a projected variant is in the schema -- the vectorized parquet reader treats it as a nested type change and throws. - HoodieFileGroupReaderBasedFileFormat passes the original Spark schemas through to the readerContext. - AvroSchemaConverterWithTimestampNTZ falls back to record conversion on unknown logical type annotations (defense-in-depth so Hudi's Avro reader path doesn't blow up if a future write produces a VARIANT(1)-annotated group). - HoodieAvroUtils.createNewSchemaField wraps Schema.Field construction in try/catch and logs the offending name on failure (diagnostic log, kept until follow-up cleanup). TestVariantDataType re-runs the read-path assertions on Spark 4.1 (INSERT, UPDATE, DELETE plus all cast(v as string) checks pass). withFixture is left in place but gated on if(false) with an updated apache#18605 comment so the test now exercises the read paths on 4.1

Spark 4.1's vectorized parquet reader path produces UnsafeRow encodings for VariantType columns that occasionally fault during shuffle copy under Hudi's pipeline (apache#18605: SIGBUS in StubRoutines::forward_copy_longs / InternalError in UnsafeRow.copy when RangePartitioner.sketch samples rows). Force row-based reading for any schema containing a VariantType column. The row-based path routes through Spark's ParquetReadSupport which materializes VariantType correctly via ParquetUnshreddedVariantConverter, eliminating the vectorized-path source of the corruption. This is a partial fix. The MERGE INTO path on Spark 4.1 remains flaky -- variant data round-tripped through Hudi's UnsafeProjection produces UnsafeRows whose variable-length encoding is unstable regardless of which read path created them, occasionally faulting during shuffle. The deeper fix is left as follow-up; restore the TestVariantDataType withFixture skip on Spark 4.1 with an updated comment naming flakiness as the remaining blocker.

…MOR merge Spark 4.1's PushVariantIntoScan rewrites a VariantType read column into a struct<0:..> with VariantMetadata, and parquet-mr does the projection natively against the base file. Log-block records, however, carry the full variant and reach the merger unaligned with the base rows, producing SIGBUS-translated InternalErrors during downstream UnsafeRow.copyMemory and (with vectorized reads disabled) wrong results. Add an engine-neutral hook on HoodieReaderContext that lets the buffer align log-block records before they hit the merger: - HoodieReaderContext.getLogBlockRecordProjection: default no-op, overridden on Spark 4.1 to project variant fields into the PushVariantIntoScan struct shape while preserving every other field of the data-block schema (including _hoodie_record_key / _tmp_metadata_row_index that the merger reads by ordinal). Projecting down to the bare required schema strips those metadata cols and the merger then reads off the end of the row. - FileGroupRecordBuffer.getProjectedTransformer composes evolution then variant projection and returns the evolved data-block schema (not readerSchema), since the projector preserves the field shape. - PositionBasedFileGroupRecordBuffer.processDataBlock and FileGroupRecordBuffer.getRecordsIterator route through the new helper. - SparkFileFormatInternalRowReaderContext.getFileRecordIterator builds the base-file read schema from the augmented requiredSchema (so _hoodie_record_key is read for key-based merge) and overlays variant projection metadata from sparkRequiredSchema (so parquet-mr still does the native variant projection). Drops the lazy variantLogProjector, which would double-project the parquet log-block path and fail buildVariantProjector's VARIANT precondition. - TestVariantDataType: re-enable on Spark 4.1.

… tests Two issues in toHoodieTypeNested were forcing SparkAdapterSupport to resolve a version-specific adapter class for any StructType conversion, breaking TestHoodieSparkSchemaUtils with ClassNotFoundException for the running profile's adapter (Spark3_5Adapter / Spark4_1Adapter) — neither ships on the hudi-spark-common test classpath: - The PushVariantIntoScan pattern guard called sparkAdapter.isVariantProjectionStruct unconditionally on every StructType. Hoist into a helper that short-circuits on Spark < 4.1 (no PushVariantIntoScan) and on structs whose fields carry no Spark Metadata (the projection always tags every field with VariantMetadata, so an empty-metadata struct is definitively not one). Catch ClassNotFoundException/NoClassDefFoundError as a belt-and-suspenders fallback for environments that still hit the adapter lookup. - The generic-StructType branch had a LOG.warn that interpolated sparkAdapter.isVariantType(st) — debug instrumentation that also triggered the adapter load. Drop it, along with the now-unused LOG / Logger / LoggerFactory.

…ad paths apache#18334 moved Spark 4.0's [value, metadata] variant reorder out of the shared HoodieParquetReadSupport into the version-specific Spark40HoodieParquetReadSupport, but two consumers still constructed the base ReadSupport directly. With no reorder, parquet-mr hands [metadata, value] bytes to Spark 4.0's ParquetUnshreddedVariantConverter (whose converters array indexes by position assuming [value, metadata]), the bytes are swapped, and Variant.<init> raises MALFORMED_VARIANT during read. Two callers fixed: - Spark40LegacyHoodieParquetFileFormat: switch the parquet-mr fallback branch to Spark40HoodieParquetReadSupport. Affects MERGE INTO and any other query that takes the legacy file format path on Spark 4.0. - HoodieSparkParquetReader (shared, used for parquet log-block reads on every Spark version): route through a new SparkAdapter.createParquetReadSupport factory whose default returns the base class and which Spark4_0Adapter overrides to return the Spark40 subclass. Affects parquet log blocks containing variant columns on Spark 4.0.

- HoodieAvroUtils: drop diagnostic try/catch + LOG.error from createNewSchemaField(.., Order); remove now-unused slf4j Logger imports and LOG field. - HoodieFileGroupReaderBasedFileFormat: demote three logInfo debug calls (entry log + per-conversion logs) to logDebug; they fire on every reader build and would flood production logs at INFO. Same treatment for the isSplitable summary log. - HoodieFileGroupReaderBasedFileFormat: gate the variant-disables-vectorized-reads branch on HoodieSparkUtils.gteqSpark4_1. Spark 4.0 reads variant via Spark40HoodieParquetReadSupport's row-based path correctly and never had the UnsafeRow shuffle-copy crash; disabling vectorized on 4.0 is a perf regression with no benefit. - FileGroupRecordBuffer: skip the optional log-block record projection when a custom payload class is configured. PayloadUpdateProcessor.handleNonDeletes round-trips through convertToAvroRecord(record, recordSchema), and the schema still describes variant fields as VariantType — feeding it a row whose variant has been rewritten into the projected struct would corrupt the avro record. Standard mergers only touch metadata cols by ordinal and are unaffected. - PositionBasedFileGroupRecordBuffer: rename schemaTransformerWithEvolvedSchema -> projectedTransformer to match the helper it now calls (getProjectedTransformer composes evolution + variant projection); rename the loop's evolved intermediate accordingly. - SparkFileFormatInternalRowReaderContext: drop unused mappedClosable companion helper (the only caller, the direct log-file projector path, was removed earlier in the PR). Also drop the now-unused CloseableMappingIterator import. - SparkFileFormatInternalRowReaderContext: in getLogBlockRecordProjection and getFileRecordIterator, look up Spark required-schema fields via a new findFieldByName helper that honors `spark.sql.caseSensitive`. Same-name lookups previously used `==` which would mis-resolve under case-insensitive analysis. - Spark4_1Adapter.buildVariantProjector: drop unused .zipWithIndex pair allocation. Replace the two raw sparkDataSchema.fieldIndex call sites with a lookupDataField helper that throws an IllegalStateException naming both schemas instead of the bare IllegalArgumentException Spark would raise — easier to triage if the projector is ever fed mismatched schemas. - AvroSchemaConverterWithTimestampNTZ: log a debug line when a group's LogicalTypeAnnotation is unrecognized and we fall back to record conversion. The fallback is correct for the variant binary group we write, but a future unknown annotation that shouldn't be a record would otherwise be silently misinterpreted.

Keep the WHY (non-obvious invariant or trade-off) and drop the WHAT (restating the code or replaying the bug investigation) across the files touched by this PR. No code changes.

…) 1) Override applyVariantLogicalType to apply LogicalTypeAnnotation.variantType((byte) 1), matching parquet 1.16+'s SparkToParquetSchemaConverter convention. Validated end-to-end through TestVariantDataType (cow + mor: insert / update / delete / merge into / select with cast(v as string)). The avro read path stays correct via AvroSchemaConverterWithTimestampNTZ's record-conversion fallback for unrecognized annotations (the variant binary group is genuinely a 2-field record).

Both getLogBlockRecordProjection and getFileRecordIterator were running the same fields.map { findFieldByName + isVariantProjectionStruct + copy(dataType) } pattern to overlay projected variant types onto a base StructType. Extracted into overlayVariantProjections so the two sites stay in sync.

Match the variable name PositionBasedFileGroupRecordBuffer uses for the same Pair<Function, HoodieSchema> return value.

hudi-agent

🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.

Thanks for working on this! The PR re-enables variant data tests on Spark 4.1 and aligns the MOR log path with PushVariantIntoScan, while restoring Spark 4.0 variant reads via a dedicated subclass. One edge case worth double-checking in the inline comment around the custom-payload-class gating — the early-return only covers the log path, but base records get projected via parquet-mr's overlay and may still hit convertToAvroRecord with a mismatching schema. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. A few small naming and idiom nits below.

hudi-agent · 2026-05-14T19:22:38Z

+   */
+  protected Pair<Function<T, T>, HoodieSchema> getProjectedTransformer(HoodieDataBlock dataBlock) {
+    Pair<Function<T, T>, HoodieSchema> evolved = getSchemaTransformerWithEvolvedSchema(dataBlock);
+    if (payloadClasses.isPresent()) {


🤖 I wonder if this payloadClasses.isPresent() guard is complete. It skips variant projection for log records, but base records read in SparkFileFormatInternalRowReaderContext.getFileRecordIterator still get the overlaid schema (struct shape) unconditionally. Later in HoodieAvroRecordMerger.merge, recordContext.convertToAvroRecord(older.getRecord(), getSchemaFromBufferRecord(older)) runs on the base record with readerSchema typing the field as VariantType — the same mis-decode the comment above warns about, but on the base side. Could you confirm there's an upstream check that prevents PushVariantIntoScan from running with PAYLOAD_BASED_MERGE_STRATEGY_UUID, or otherwise gate the overlay similarly? @yihua

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

hudi-agent · 2026-05-14T19:22:38Z

      supportVectorizedRead = false
      supportReturningBatch = false
      false
+    } else if (schema.fields.exists(f => f.dataType.isInstanceOf[StructType]


🤖 nit: could you use a pattern match instead of the isInstanceOf/asInstanceOf pair? Something like schema.fields.exists(_.dataType match { case st: StructType => sparkAdapter.isVariantProjectionStruct(st); case _ => false }) is more idiomatic Scala and avoids the double-cast.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

hudi-agent · 2026-05-14T19:22:38Z

+                                       datetimeRebaseSpec: RebaseSpec,
+                                       int96RebaseSpec: RebaseSpec,
+                                       tableSchemaOpt: org.apache.hudi.common.util.Option[org.apache.parquet.schema.MessageType] = org.apache.hudi.common.util.Option.empty())
+  extends HoodieParquetReadSupport(


🤖 nit: could you add an import alias (e.g. import org.apache.hudi.common.util.{Option => HOption}) and use HOption here, rather than the fully-qualified name inline? That matches the convention used in SparkFileFormatInternalRowReaderContext.scala and similar files in this module.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

hudi-agent · 2026-05-14T19:22:38Z

+  // SparkToParquetSchemaConverter convention.
+  override protected def applyVariantLogicalType(builder: Types.GroupBuilder[GroupType]): Types.GroupBuilder[GroupType] = {
+    builder.as(LogicalTypeAnnotation.variantType(1.toByte))
+  }


🤖 nit: the 1 is the Variant spec version — it might be worth extracting it as a named constant (e.g. private val VariantSpecVersion: Byte = 1) so a future reader doesn't have to cross-reference the parquet spec to understand what the literal represents.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

yihua · 2026-05-14T19:20:25Z

+  def buildVariantProjector(sparkDataSchema: StructType,
+                            sparkRequiredSchema: StructType): Option[InternalRow => InternalRow] = None


We need to revisit if this impacts performance and see if such projection can be avoided through custom variant reader.

#18739 to follow up

yihua · 2026-05-14T19:29:39Z

+  private def isVariantGroup(group: GroupType): Boolean = {
+    group.containsField("value") &&
+      group.containsField("metadata") &&
+      group.getType("value").isPrimitive &&
+      group.getType("metadata").isPrimitive &&
+      group.getType("value").asPrimitiveType().getPrimitiveTypeName == PrimitiveType.PrimitiveTypeName.BINARY &&
+      group.getType("metadata").asPrimitiveType().getPrimitiveTypeName == PrimitiveType.PrimitiveTypeName.BINARY
+  }


nit: can this match non-variant GroupType with the same schema?

We can revisit this later. This moves existing code only.

yihua · 2026-05-14T19:51:18Z

+      case Some(p) => HOption.of(new JFunction[InternalRow, InternalRow] {
+        // .copy() because the buffer stores rows into ExternalSpillableMap and
+        // UnsafeProjection reuses a single output buffer.
+        override def apply(r: InternalRow): InternalRow = p(r).copy()


@voonhous Could you check why copy() is required? The log record reading should already have done the record copy when putting the record into the log record map.

Caller no longer threads the fixed LEGACY int96 spec; the adapter sets it internally so the public signature is one parameter shorter and the hardcoded policy lives in one place. Also cleans up import aliases (HOption, MessageType, RebaseSpec) so the signature reads without package prefixes.

- Remove sparkDataSchema parameter from SparkFileFormatInternalRowReaderContext (no in-class reference) and its sole call site. - Flip the double-negative !exists(!= empty) to forall(== empty) for readability in isSparkVariantProjectionStruct.

The shape-only isVariantGroup heuristic can false-positive on a user-defined struct<value: binary, metadata: binary>, silently swapping its fields. Read the Spark catalyst requested schema from ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA and only reorder top-level fields whose catalyst type is actually VariantType. Falls back to the shape-only check when the catalyst schema is unavailable, preserving the existing TestSpark40HoodieParquetReadSupport.testReorderVariantFieldsNonVariantGroupsUnchanged behavior. Also cleans up imports (HOption alias, bare MessageType).

yihua

LGTM

hudi-bot · 2026-05-14T21:07:24Z

CI report:

27e0a10 Azure: SUCCESS
601254b Azure: PENDING

Bot commands

@hudi-bot supports the following commands:

@hudi-bot run azure re-run the last Azure build

codecov-commenter · 2026-05-14T22:22:51Z

Codecov Report

❌ Patch coverage is 78.80795% with 32 lines in your changes missing coverage. Please review.
✅ Project coverage is 68.14%. Comparing base (0d5b442) to head (601254b).
⚠️ Report is 2 commits behind head on master.

Files with missing lines	Patch %	Lines
...rces/parquet/Spark40HoodieParquetReadSupport.scala	78.94%	1 Missing and 7 partials ⚠️
...org/apache/spark/sql/adapter/Spark4_1Adapter.scala	71.42%	4 Missing and 4 partials ⚠️
...hudi/SparkFileFormatInternalRowReaderContext.scala	81.48%	2 Missing and 3 partials ⚠️
...parquet/HoodieFileGroupReaderBasedFileFormat.scala	71.42%	3 Missing and 1 partial ⚠️
...e/spark/sql/avro/HoodieSparkSchemaConverters.scala	50.00%	1 Missing and 2 partials ⚠️
...g/apache/spark/sql/adapter/BaseSpark4Adapter.scala	60.00%	0 Missing and 2 partials ⚠️
...scala/org/apache/spark/sql/hudi/SparkAdapter.scala	75.00%	1 Missing ⚠️
...parquet/Spark40LegacyHoodieParquetFileFormat.scala	0.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             master   #18674    +/-   ##
==========================================
  Coverage     68.13%   68.14%            
- Complexity    29044    29095    +51     
==========================================
  Files          2516     2517     +1     
  Lines        141003   141113   +110     
  Branches      17487    17508    +21     
==========================================
+ Hits          96076    96159    +83     
- Misses        37035    37046    +11     
- Partials       7892     7908    +16

Flag	Coverage Δ
common-and-other-modules	`44.41% <16.00%> (-0.01%)`	⬇️
hadoop-mr-java-client	`45.00% <75.00%> (-0.02%)`	⬇️
spark-client-hadoop-common	`48.32% <29.50%> (-0.03%)`	⬇️
spark-java-tests	`48.98% <51.65%> (-0.02%)`	⬇️
spark-scala-tests	`44.91% <73.50%> (+0.01%)`	⬆️
utilities	`37.62% <54.66%> (-0.01%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...ache/hudi/io/storage/HoodieSparkParquetReader.java	`78.49% <100.00%> (ø)`
...datasources/parquet/HoodieParquetReadSupport.scala	`94.44% <100.00%> (+1.85%)`	⬆️
...apache/hudi/common/engine/HoodieReaderContext.java	`96.42% <100.00%> (+0.03%)`	⬆️
...ommon/table/read/buffer/FileGroupRecordBuffer.java	`95.00% <100.00%> (+0.31%)`	⬆️
...ead/buffer/PositionBasedFileGroupRecordBuffer.java	`75.00% <100.00%> (ø)`
...quet/avro/AvroSchemaConverterWithTimestampNTZ.java	`76.47% <100.00%> (+0.35%)`	⬆️
...org/apache/spark/sql/adapter/Spark4_0Adapter.scala	`64.17% <100.00%> (+1.67%)`	⬆️
...ion/datasources/parquet/Spark40ParquetReader.scala	`95.62% <100.00%> (ø)`
...scala/org/apache/spark/sql/hudi/SparkAdapter.scala	`60.00% <75.00%> (+5.45%)`	⬆️
...parquet/Spark40LegacyHoodieParquetFileFormat.scala	`0.00% <0.00%> (ø)`
... and 6 more

... and 18 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

This PR was opened on 2026-04-27. Several fixes have since landed on master that targeted hudi-spark4.0.x / hudi-spark4.1.x but were not picked up by rebases of this branch. Mirror them into hudi-spark4.2.x so 4.2 starts at parity with 4.1: - Spark4_2Adapter.getDateTimeRebaseMode: honor SparkSession overrides (apache#18675). - Spark4_2Adapter: override isVariantProjectionStruct / buildVariantProjector / applyVariantLogicalType so variant push-down works on 4.2 (apache#18674). - Spark42LegacyHoodieParquetFileFormat: use HoodieStorageUtils factory so a user-configured hoodie.storage.class is honored (apache#17661). - HoodieSpark4_2ExtendedSqlAstBuilder + SqlBase.g4: accept VECTOR(dim, elementType) DDL syntax (apache#18488 / apache#18779). - AvroDeserializer: add Hudi customization comment block documenting the Avro 1.12.1 fast-reader logical-type handling (apache#18773). - Add TestSpark4_2AvroLogicalTypeBytes parallel to the 4.1 test (apache#18773). - Fix "transfering" -> "transferring" typo in two parquet read paths (apache#18763).

…restore Spark 4.0 reads (#18674) Co-authored-by: Y Ethan Guo <ethan.guoyihua@gmail.com>

…restore Spark 4.0 reads (apache#18674) Co-authored-by: Y Ethan Guo <ethan.guoyihua@gmail.com>

This PR was opened on 2026-04-27. Several fixes have since landed on master that targeted hudi-spark4.0.x / hudi-spark4.1.x but were not picked up by rebases of this branch. Mirror them into hudi-spark4.2.x so 4.2 starts at parity with 4.1: - Spark4_2Adapter.getDateTimeRebaseMode: honor SparkSession overrides (apache#18675). - Spark4_2Adapter: override isVariantProjectionStruct / buildVariantProjector / applyVariantLogicalType so variant push-down works on 4.2 (apache#18674). - Spark42LegacyHoodieParquetFileFormat: use HoodieStorageUtils factory so a user-configured hoodie.storage.class is honored (apache#17661). - HoodieSpark4_2ExtendedSqlAstBuilder + SqlBase.g4: accept VECTOR(dim, elementType) DDL syntax (apache#18488 / apache#18779). - AvroDeserializer: add Hudi customization comment block documenting the Avro 1.12.1 fast-reader logical-type handling (apache#18773). - Add TestSpark4_2AvroLogicalTypeBytes parallel to the 4.1 test (apache#18773). - Fix "transfering" -> "transferring" typo in two parquet read paths (apache#18763).

…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.

…el projection hook (#18923) * refactor(reader): push variant projection into log readers, drop buffer hook Remove the engine-neutral FileGroupRecordBuffer variant-projection composition (#18674's getLogBlockRecordProjection hook) so the merge buffer stays format-agnostic; each log reader now emits rows already aligned to the projected read schema (#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 #18739. * refactor(variant): address review nits on log-block variant projection - HoodieSparkParquetReader: rename variant-overload parameter to structSchema and drop the no-op alias. - SparkFileFormatInternalRowReaderContext: extract isPayloadBased and drop the double negation in shouldProjectVariants. * refactor(variant): address review nits on parquet reader param and merger guard - Rename getUnsafeRowIterator param structSchema to projectedStructSchema - Add inline comment explaining the merger != null guard * review: address nit

voonhous changed the title ~~fix: Fix #18671 (will amend title later - triggering CI now)~~ May 1, 2026

voonhous requested review from rahil-c and yihua and removed request for yihua May 1, 2026 21:46

voonhous force-pushed the fix-#18605 branch from b7d7c85 to add9698 Compare May 1, 2026 21:51

github-actions Bot added the size:L PR with lines of changes in (300, 1000] label May 1, 2026

hudi-agent reviewed May 1, 2026

View reviewed changes

hudi-agent reviewed May 2, 2026

View reviewed changes

voonhous changed the title ~~fix: Fix #18605 (will amend title later - triggering CI now)~~ May 2, 2026

voonhous force-pushed the fix-#18605 branch from 8ccd3f4 to 4b33bc9 Compare May 2, 2026 12:57

hudi-agent reviewed May 2, 2026

View reviewed changes

Comment thread ...di-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala

Comment thread hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/FileGroupRecordBuffer.java Outdated

voonhous force-pushed the fix-#18605 branch from 9002f9b to 9ef0b22 Compare May 2, 2026 17:04

hudi-agent reviewed May 2, 2026

View reviewed changes

voonhous added this to the release-1.2.0 milestone May 13, 2026

voonhous added 11 commits May 14, 2026 12:07

feat: address todo apache#18334

1e49ec0

docs: trim verbose comments added in this PR

eb3ea5e

Keep the WHY (non-obvious invariant or trade-off) and drop the WHAT (restating the code or replaying the bug investigation) across the files touched by this PR. No code changes.

refactor: rename projected -> projectedTransformer in getRecordsIterator

27e0a10

Match the variable name PositionBasedFileGroupRecordBuffer uses for the same Pair<Function, HoodieSchema> return value.

yihua force-pushed the fix-#18605 branch from 9ef0b22 to 27e0a10 Compare May 14, 2026 19:10

hudi-agent reviewed May 14, 2026

View reviewed changes

yihua reviewed May 14, 2026

View reviewed changes

yihua mentioned this pull request May 14, 2026

Revisit projection for variant type columns on Spark #18739

Closed

5 tasks

yihua added 3 commits May 14, 2026 13:58

yihua approved these changes May 14, 2026

View reviewed changes

yihua merged commit 1b26f2b into apache:master May 14, 2026
62 of 63 checks passed

yihua added the engine:spark4.1 Spark 4.1 features and improvements label May 19, 2026

yihua added a commit that referenced this pull request May 20, 2026

fix(variant): align Spark 4.1 MOR merge with PushVariantIntoScan and …

2929f7d

…restore Spark 4.0 reads (#18674) Co-authored-by: Y Ethan Guo <ethan.guoyihua@gmail.com>

dwshmilyss pushed a commit to dwshmilyss/hudi that referenced this pull request May 21, 2026

fix(variant): align Spark 4.1 MOR merge with PushVariantIntoScan and …

7b21550

…restore Spark 4.0 reads (apache#18674) Co-authored-by: Y Ethan Guo <ethan.guoyihua@gmail.com>

yihua mentioned this pull request May 28, 2026

docs: Update documentation for new features in Hudi 1.2.0 #18867

Merged

3 tasks

voonhous deleted the fix-#18605 branch June 4, 2026 05:07

This was referenced Jun 5, 2026

Spark 4.0.x reads Variant Parquet group fields by position instead of by name, causing MALFORMED_VARIANT #18334

Closed

refactor(variant): self-align log-block variant rows, drop buffer-level projection hook #18923

Merged

voonhous mentioned this pull request Aug 27, 2026

[BUG] Payload-based MOR tables mis-decode variant rows once Spark 4.1 PushVariantIntoScan fires #19768

Open

		def buildVariantProjector(sparkDataSchema: StructType,
		sparkRequiredSchema: StructType): Option[InternalRow => InternalRow] = None

Uh oh!

Conversation

voonhous commented May 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Describe the issue this Pull Request addresses

Summary and Changelog

Impact

Risk Level

Documentation Update

Contributor's checklist

hudi-agent left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

hudi-agent left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

hudi-agent left a comment

Choose a reason for hiding this comment

hudi-agent left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

hudi-agent left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hudi-agent left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yihua left a comment

Choose a reason for hiding this comment

hudi-bot commented May 14, 2026

CI report:

Uh oh!

codecov-commenter commented May 14, 2026

Codecov Report

Labels

5 participants

voonhous commented May 1, 2026 •

edited

Loading