voonhous opened a new issue, #20307:
URL: https://github.com/apache/hudi/issues/20307
**Describe the problem you faced**
A COW upsert fails when the base file holds a shredded variant column with
at least one DECIMAL-typed shredded leaf. The bulk_insert that writes the
shredded file succeeds and Spark reads it back correctly; the merge that
rewrites the file group fails in the Avro read path with
```
org.apache.hudi.exception.HoodieUpsertException: Failed to compact file
group: 6213365d-...
at
org.apache.hudi.io.FileGroupReaderBasedMergeHandle.doMerge(FileGroupReaderBasedMergeHandle.java:337)
Caused by: org.apache.hudi.exception.HoodieException: unable to read next
record from parquet file
at
org.apache.hudi.common.util.ParquetReaderIterator.hasNext(ParquetReaderIterator.java:65)
...
at
org.apache.hudi.io.FileGroupReaderBasedMergeHandle.doMerge(FileGroupReaderBasedMergeHandle.java:297)
Caused by: org.apache.hudi.exception.HoodieAvroSchemaException: Failed to
parse schema: {"type":"record","name":"hudi_record",...
at
org.apache.hudi.common.schema.HoodieSchema$Parser.parse(HoodieSchema.java:1713)
at org.apache.hudi.common.schema.HoodieSchema.parse(HoodieSchema.java:375)
at
org.apache.parquet.avro.HoodieAvroReadSupport.init(HoodieAvroReadSupport.java:90)
Caused by: org.apache.avro.AvroTypeException: Undefined schema:
k102.typed_value
at
org.apache.avro.util.SchemaResolver$ResolvingVisitor.visitNonTerminal(SchemaResolver.java:187)
...
at org.apache.avro.Schema$Parser.parse(Schema.java:1516)
```
Any JSON number with a fractional part is enough to hit this:
`parse_json('{"a":1.5}')` stores the value as a decimal, and the inferred
shredding schema writes it as `fixed_len_byte_array(8) typed_value
(DECIMAL(18,12))`. Unshredded variant columns, and shredded columns whose
leaves are all string/int/long/boolean/array, upsert fine.
**Cause**
`HoodieAvroParquetReader.getIndexedRecordIteratorInternal` reads a shredded
base file in its shredded shape
(`HoodieVariantReconstruction.intermediateSchema()`, from #18938). That schema
is built by `VariantSchemaUtils.toShreddedReadSchema(requestedSchema,
fileSchema)`, where `fileSchema` is the parquet footer converted by
parquet-avro's `AvroSchemaConverter` (`NativeAvroSchemaConverter.java:41`).
parquet-avro names the Avro `fixed` for a DECIMAL `fixed_len_byte_array`
leaf after its parquet field, so the leaf becomes a `fixed` called
`typed_value`. The enclosing shredded group also converts to a `record` called
`typed_value`. Two named types now share the fullname `typed_value`.
The reader then calls `AvroReadSupport.setRequestedProjection(conf,
schema.toAvroSchema())` (`HoodieAvroParquetReader.java:199` and `:203`), which
stores `schema.toString()`. Avro's `toString` writes each named type once and
emits later occurrences as a bare name reference. `HoodieAvroReadSupport.init`
re-parses that string (`HoodieAvroReadSupport.java:90`); the bare
`"typed_value"` under key `k10` is resolved in that record's namespace (`k102`,
from parquet-avro's duplicate-name suffixing), where nothing is defined.
Where the enclosing namespace happens to be the empty one, the same
reference resolves to the `typed_value` record instead of the fixed, so the
non-throwing form of this is a silent wrong type rather than a failure.
The dumped schema string shows the shape; the DECIMAL leaves are the only
ones emitted as references:
```
{"name":"k10","type":{"type":"record","name":"k10","namespace":"k102","fields":[
{"name":"value","type":["null","bytes"],"default":null},
{"name":"typed_value","type":["null","typed_value"],"default":null}]}}
```
No `fixed` named `typed_value` is defined anywhere in the string.
**To Reproduce**
Verified on master `a41f6ea` with the `-Pspark4.1` bundle (Spark 4.1.1,
Scala 2.13), COW, Spark datasource writes, default configs
(`hoodie.parquet.variant.write.shredding.enabled=true`, inference on):
1. bulk_insert a DataFrame with a `payload` column produced by `parse_json`
over JSON documents that contain at least one number with a fractional part,
e.g. `{"k3": 0.4375, "k7": "x"}`.
2. upsert a batch that updates some of those keys.
Step 2 fails every time with the trace above. The same input with
`hoodie.parquet.variant.write.shredding.enabled=false` on both writes succeeds,
and a Spark read of the table written by step 1 returns the correct rows and
`variant_get` values.
A one-row SQL form (`insert ... parse_json('{"a":1.5}')` then `update`) is
expected to reproduce it as well; the verified run used a 20000-row generated
workload.
**Expected behavior**
The upsert succeeds and the merged file keeps the shredded layout (or any
layout, as long as the values round-trip).
**Environment Description**
* Hudi version : 1.3.0-SNAPSHOT (master `a41f6ea`)
* Spark version : 4.1.1
* Storage : local filesystem
* Table type : COPY_ON_WRITE
**Additional context**
Existing shredded-variant coverage (`TestVariantShreddingMixedLayouts`,
`TestVariantDataType`) only uses integer and string leaves, which is why the
Avro-path merge never saw a `fixed` leaf. MOR compaction reads base files
through the same reader and is probably affected too; not run.
Fix options: give the DECIMAL `fixed` a path-unique name during footer
conversion, or stop round-tripping the projection schema through `toString` in
`HoodieAvroParquetReader`.
Related: #18938 (shredded base files through the Avro reader), #18961
(per-file shredding inference on by default, which makes the DECIMAL leaf
appear without a forced schema).
--
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]
