voonhous opened a new issue, #20442: URL: https://github.com/apache/hudi/issues/20442
The same variant DataFrame was written through vanilla Spark, Hudi and Iceberg on one Spark 4.1.1 build, unshredded and shredded (each engine's default schema inference), and the parquet footers were compared. **Summary** - **Unshredded**: the `metadata` and `value` bytes are identical across all writers on every row. - **Shredded**: the `metadata` bytes are identical across all writers, and each engine reads its own files back to the input. Hudi matches vanilla Spark except for two spec deviations that are already known (D1, D2). The other differences come from choices made by Iceberg or Spark. - Not covered yet: reading each engine's files with the other engines. Versions: Hudi master f39f99713f9f (hudi-spark4.1-bundle_2.13 1.3.0-SNAPSHOT), Spark 4.1.1, iceberg-spark-runtime-4.1_2.13 1.12.0, JDK 17. Hudi was written three ways: COW `bulk_insert` (row writer), and COW `insert` with the SPARK and with the AVRO record type. Spec = Parquet [VariantShredding.md](https://github.com/apache/parquet-format/blob/master/VariantShredding.md). ### Differences | id | mode | what | vanilla Spark | Hudi | Iceberg | spec | Hudi side | |---|---|---|---|---|---|---|---| | D1 | shredded | decimal `typed_value` | INT64 DECIMAL(18,s) | **FIXED_LEN_BYTE_ARRAY(8) DECIMAL(18,s)**, all three write paths | INT32 DECIMAL(p,s), narrowest precision seen | decimal8 must be INT64 | deviation, already in #20322 | | D2 | shredded | shredded array layout | 3-level `list` / `element` | 3-level; **AVRO record type writes 2-level `repeated group array`** | 3-level | must be 3-level | deviation, known (#19782) | | D3 | shredded | int leaf types | every int widened to INT64 | same as Spark | narrowest type seen, INT(8/16/32/64) | both valid | follows Spark | | D4 | shredded | `a` = 1, 2, 300, 4 | all four in `typed_value` | same as Spark | INT(16) leaf; 1, 2, 4 kept in `value` as variant int8, only 300 typed | both readings exist | follows Spark | | D5 | shredded | key that is always JSON null (`n`) | shredded as a group with only `value` | same as Spark | not shredded; stays in the top-level residual `value` | both valid | follows Spark | | D6 | both | footer key-value metadata | `org.apache.spark.sql.parquet.row.metadata` (Spark schema, type `variant`) | **no Spark row metadata**; Hudi bloom filter + min/max key; AVRO path adds `parquet.avro.schema` | `iceberg.schema` (type `variant`) | n/a | check when other engines read Hudi files | | D7 | unshredded | child order in the variant group | `value`, `metadata` | `metadata`, `value` | `metadata`, `value` | readers must match by name | none | | D8 | both | other | no field ids, SNAPPY, parquet-mr 1.16.0 | no field ids, ZSTD, parquet-mr 1.16.0, five `_hoodie_*` columns | field ids on every field, ZSTD, parquet-mr 1.17.1; dictionary-encodes the residual `value` | n/a | none | ### Shredded `typed_value` leaf types The three Hudi write paths produce the same leaf types; the AVRO record type differs only in list structure (D2). | leaf | vanilla Spark | Hudi | Iceberg | |---|---|---|---| | `i8`, `i16`, `i32` | INT64 | INT64 | INT32 INT(8) / INT(16) / INT(32) | | `i64` | INT64 | INT64 | INT64 INT(64) | | `f32` / `f64` | FLOAT / DOUBLE | FLOAT / DOUBLE | FLOAT / DOUBLE | | `dec4` (decimal(9,2) input) | INT64 DECIMAL(18,2) | **FLBA(8) DECIMAL(18,2)** | INT32 DECIMAL(3,2) | | `dec8` (decimal(18,4) input) | INT64 DECIMAL(18,4) | **FLBA(8) DECIMAL(18,4)** | INT32 DECIMAL(5,4) | | `dec16` (decimal(38,6) input) | INT64 DECIMAL(18,6) | **FLBA(8) DECIMAL(18,6)** | INT32 DECIMAL(7,6) | | `bool` | BOOLEAN | BOOLEAN | BOOLEAN | | `str`, `obj.b` | BYTE_ARRAY STRING | BYTE_ARRAY STRING | BYTE_ARRAY STRING | | `bin` | BYTE_ARRAY | BYTE_ARRAY | BYTE_ARRAY | | `date` | INT32 DATE | INT32 DATE | INT32 DATE | | `ts` / `ts_ntz` | INT64 TIMESTAMP(UTC / local, MICROS) | same | same | | `obj.a`, `arr[]`, `arr_obj[].k` | INT64 | INT64 | INT32 INT(8) | | `obj.inner.c` | DOUBLE | DOUBLE | DOUBLE | | `v_json.a` (1, 2, 300, 4) | INT64 | INT64 | INT32 INT(16) (see D4) | | `v_json.c` (1.25, 2.5, 3.75, 4.0) | INT64 DECIMAL(18,2) | **FLBA(8) DECIMAL(18,2)** | INT32 DECIMAL(3,2) | | `v_json.n` (always null) | value-only group | value-only group | not shredded (D5) | ### Same in every writer - The variant group is annotated `VARIANT(1)`. - The metadata dictionary is unsorted, uses 1-byte offsets and lists keys in first-seen order. It still contains the shredded keys, and its bytes are identical in both modes. - A SQL NULL makes the whole group null. A variant null (`parse_json('null')`) is stored as `value` = `00` with `typed_value` null. - Shredded `v_typed`: every writer shreds the same 18 fields, and the decoded `(value, typed_value)` of every row matches Spark. <details><summary>Input</summary> `v_typed` uses `to_variant_object` so that every leaf keeps its Spark type. `v_json` covers what `parse_json` produces: missing keys, an extra key in an array element, empty containers, mixed decimal scales, and a variant null. ```sql SELECT id, CASE WHEN id = 5 THEN NULL ELSE to_variant_object(named_struct( 'i8', CAST(id AS TINYINT), 'i16', CAST(1000 + id AS SMALLINT), 'i32', CAST(100000 + id AS INT), 'i64', CAST(10000000000 + id AS BIGINT), 'f32', CAST(id + 0.5 AS FLOAT), 'f64', CAST(id + 0.25 AS DOUBLE), 'dec4', CAST(id + 0.12 AS DECIMAL(9, 2)), 'dec8', CAST(id + 0.1234 AS DECIMAL(18, 4)), 'dec16', CAST(id + 0.123456 AS DECIMAL(38, 6)), 'bool', id % 2 = 0, 'str', concat('s', id), 'bin', CAST(concat('b', id) AS BINARY), 'date', date_add(DATE'2024-01-01', id), 'ts', TIMESTAMP'2024-01-01 00:00:00' + make_dt_interval(0, id), 'ts_ntz', TIMESTAMP_NTZ'2024-01-01 00:00:00' + make_dt_interval(0, id), 'obj', named_struct('a', id, 'b', concat('x', id), 'inner', named_struct('c', CAST(id AS DOUBLE))), 'arr', array(id, id + 1, id + 2), 'arr_obj', array(named_struct('k', id), named_struct('k', id + 1)) )) END AS v_typed, parse_json(json) AS v_json FROM src ``` | id | `json` | |---|---| | 1 | `{"a": 1, "b": "x1", "c": 1.25, "d": true, "e": {"f": [1, 2], "g": "y1"}, "h": [{"k": 1}, {"k": 2}], "n": null}` | | 2 | `{"a": 2, "b": "x2", "c": 2.5, "d": false, "e": {"f": [3], "g": "y2"}, "h": [{"k": 3}], "n": null}` | | 3 | `{"a": 300, "b": "x3", "c": 3.75, "d": true, "e": {"f": [], "g": "y3"}, "h": [], "n": null}` | | 4 | `{"a": 4, "c": 4.0, "d": false, "e": {}, "h": [{"k": 4, "extra": "z"}]}` | | 5 | `null` | </details> <details><summary>Write settings</summary> Every write produces one data file; the session time zone is UTC. | writer | write | shredding switch | |---|---|---| | vanilla Spark | `df.write.parquet` | `spark.sql.variant.writeShredding.enabled`, `spark.sql.variant.inferShreddingSchema=true` | | Hudi row writer | COW `bulk_insert`, `hoodie.bulkinsert.sort.mode=NONE` | `hoodie.parquet.variant.write.shredding.enabled` (inference on by default) | | Hudi SPARK record | COW `insert`, `hoodie.write.record.merge.custom.implementation.classes=org.apache.hudi.DefaultSparkRecordMerger` | same | | Hudi AVRO record | COW `insert`, `...=org.apache.hudi.common.model.HoodieAvroRecordMerger` | same | | Iceberg | format v3 table, `writeTo(...).append()` | table property `write.parquet.shred-variants` | Hudi: `NonpartitionedKeyGenerator`, record key `id`, `hoodie.metadata.enable=false`, shuffle parallelism 1. </details> <details><summary>D4 / D5: Iceberg residual for <code>v_json</code></summary> | id | `typed_value.a` (Spark, Hudi) | `typed_value.a` (Iceberg) | top-level `value` (Spark, Hudi) | top-level `value` (Iceberg) | |---|---|---|---|---| | 1 | typed 1 | value `0c01` (int8 1) | null | object with field `n` | | 2 | typed 2 | value `0c02` | null | object with field `n` | | 3 | typed 300 | typed 300 | null | object with field `n` | | 4 | typed 4 | value `0c04` | null | null | | 5 | null | null | `00` (variant null) | `00` | </details> <details><summary>D2: AVRO record type list layout</summary> ```diff optional group typed_value (LIST) { - repeated group list { - required group element { - optional binary value; - optional int64 typed_value; - } + repeated group array { + optional binary value; + optional int64 typed_value; } } ``` </details> <details><summary>Files</summary> Attached: `variant-parity-2026-10-27.zip` contains the ten parquet files (`parquet/<mode>/<writer>.parquet`), the full generated report (parquet-java schemas, leaf/encoding matrices, per-row metadata and value bytes), `footers.json`, the input, and each engine's read-back. </details> -- 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]
