voonhous opened a new issue, #20157:
URL: https://github.com/apache/hudi/issues/20157
### Describe the problem you faced
On Spark 4.0 with default confs, a variant nested inside a struct (`s
struct<inner: variant>`) breaks once the table is updated. A top-level variant
in the same table is not affected, and reads right after the insert are correct.
- **MOR, either record merger:** once an update puts a log file on the file
slice, any read of `s.inner` fails with `[MALFORMED_VARIANT]`. The base and log
files on disk are correct.
- **COW with the Spark record merger:** the update rewrites the base file,
and the rows carried over from the old file are written with the `metadata` and
`value` bytes of `s.inner` swapped. Every later read of those rows fails with
`[MALFORMED_VARIANT]`. The updated rows are written correctly.
- **COW with the Avro record merger** (the default) is correct.
Spark 4.1+ is not affected.
### To Reproduce
```sql
-- needed for the COW leg only
set
hoodie.write.record.merge.custom.implementation.classes=org.apache.hudi.DefaultSparkRecordMerger;
set hoodie.logfile.data.block.format=parquet;
create table t (id int, v variant, s struct<inner: variant>, ts long) using
hudi
tblproperties (primaryKey = 'id', preCombineField = 'ts', type = 'cow',
hoodie.compact.inline = 'false')
location '...';
insert into t select cast(id as int) as id,
parse_json(concat('{"k":"x', id, '"}')) as v,
named_struct('inner', parse_json(concat('{"k":"x', id, '"}'))) as s,
1000L as ts from range(0, 5, 1, 1);
select id, variant_get(s.inner, '$.k', 'string') from t order by id; --
correct
update t set s = named_struct('inner', parse_json(concat('{"k":"y', id,
'"}'))), ts = 1001 where id >= 3;
select id, variant_get(s.inner, '$.k', 'string') from t order by id; --
[MALFORMED_VARIANT]
select id, s from t order by id; --
[MALFORMED_VARIANT]
select id, variant_get(v, '$.k', 'string') from t order by id; --
correct
```
With `type = 'mor'` the same two statements fail with or without the two
`set` lines.
<details>
<summary>Bytes on disk after the COW update</summary>
First byte of each binary, read as plain parquet by column name (`s
struct<inner: struct<metadata: binary, value: binary>>`). A variant metadata
binary starts with the version byte `01`, an object value with `02`.
```
base file written by the insert: id 0-4 v(meta=01,value=02)
s.inner(meta=01,value=02)
base file written by the update: id 0-2 v(meta=01,value=02)
s.inner(meta=02,value=01)
id 3-4 v(meta=01,value=02)
s.inner(meta=01,value=02)
```
Both files declare the groups as `{ required binary metadata; required
binary value; }`. On MOR the base file and the log file both hold
`meta=01,value=02` for every row.
</details>
<details>
<summary>Stack trace</summary>
```
org.apache.spark.SparkRuntimeException: [MALFORMED_VARIANT] Variant binary
is malformed. Please check the data source is valid. SQLSTATE: 22023
at
org.apache.spark.types.variant.VariantUtil.malformedVariant(VariantUtil.java:179)
at org.apache.spark.types.variant.Variant.<init>(Variant.java:62)
at org.apache.spark.types.variant.Variant.<init>(Variant.java:53)
at
org.apache.spark.sql.catalyst.expressions.variant.VariantGet$.variantGet(variantExpressions.scala:379)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
Source)
```
</details>
### Suspected cause
<details>
<summary>Top-level-only reorder in the Spark 4.0 read shim (not confirmed by
a fix)</summary>
Spark 4.0's `ParquetUnshreddedVariantConverter` reads the variant group by
position as `[value, metadata]` (#18334), while Hudi writes `[metadata,
value]`. `Spark40HoodieParquetReadSupport.reorderVariantFields` works around
that by reordering the requested group, but when it is handed the catalyst
schema it reorders only top-level `VariantType` fields. A variant below a
struct keeps the file order on the row-based path, which is the path the file
group reader and the Spark-record merge handle read through.
</details>
### Expected behavior
The same rows before and after the update, and no swapped bytes on disk.
### Environment Description
* Hudi version : master `4d3af133b42e`. The same three legs fail on the
`release-1.2.1` tag and on branch-1.2.x `eb1213214ef3`; the bytes on disk were
inspected on master only.
* Spark version : 4.0.2 (Scala 2.13, JDK 17), local
* Table type : COW and MOR
### Additional context
Master CI builds no Spark 4.0 profile. Related: #18334, #18674, #18935.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
