Skip to content

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

Description

@voonhous

Describe the problem you faced

Split out of the #18923 review (#18923 (comment)). On a MOR table whose merger is payload-based (merge mode CUSTOM with a user payload class, strategy 00000000-0000-0000-0000-000000000000), a Spark 4.1+ query that lets PushVariantIntoScan rewrite a variant column (for example cast(v as string)) reads base rows in the projected struct shape but log rows as plain VariantVal, and the two shapes meet in the avro round trip the payload merger performs:

  • base + log for the same key: HoodieAvroRecordMerger.merge converts the base row to avro against a schema that still types the column as VARIANT, so the projected struct bytes get decoded as a variant.
  • log-only key: PayloadUpdateProcessor.handleNonDeletes rewrites the record through avro and hands back a VariantVal-shaped row, which HoodieFileGroupReaderBasedFileFormat.appendPartitionAndProject then projects as if it were the struct.

Traced from the code, not yet reproduced. It predates #18923: the base-side overlay in SparkFileFormatInternalRowReaderContext.getFileRecordIterator has been unconditional since #18674, and the payload skip on the log side (payloadClasses.isPresent() in the old FileGroupRecordBuffer.getProjectedTransformer, shouldProjectVariants() after #18923) only stops the log rows from being rewritten. The same gap was pointed out on #18674 (#18674 (comment)) and never picked up.

To Reproduce

Expected trigger, unrun. Spark SQL under the spark4.1 or spark4.2 profile (Spark 4.0 has no PushVariantIntoScan):

create table t (id int, v variant, ts long) using hudi
 location '/tmp/variant_payload_repro'
 tblproperties (
  primaryKey = 'id', type = 'mor', preCombineField = 'ts',
  payloadClass = 'com.example.MyPayload'
 );
-- any payload class outside the built-in set lands the table on
-- hoodie.record.merge.mode=CUSTOM and the payload-based strategy id
insert into t values (1, parse_json('{"key":"value1"}'), 1000), (2, parse_json('{"key":"value2"}'), 1000);
update t set v = parse_json('{"key":"updated"}') where id = 1;
select id, cast(v as string) from t order by id;

Expected behavior

Variant reads on payload-based MOR tables return the same values as on the default mergers, whether or not the query lets Spark push the variant extraction into the scan.

Possible direction

Both sides of the merge and the reader output have to agree on one shape. For payload-based tables that probably means reading base and log rows as VariantVal, letting the avro round trip see the schema it expects, and applying the PushVariantIntoScan projection once on the reader output instead of inside the scan. Gating the base overlay on shouldProjectVariants() alone is not enough, because appendPartitionAndProject still expects the projected shape.

Environment Description

  • Hudi version: master (1.3.0-SNAPSHOT)
  • Spark version: 4.1 / 4.2 (PushVariantIntoScan)
  • Storage: any
  • Running on Docker: no

Metadata

Metadata

Assignees

No one assigned

    Labels

    type:bugBug reports and fixes

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions