Re: [I] Flink inference of Hudi unstructured logical types (Variant, Blob, Vector) from Parquet [hudi]
kbuci commented on issue #18711: URL: https://github.com/apache/hudi/issues/18711#issuecomment-4427145763 Thanks for sharing - I noticed while updating my PR that those changes are already there. But I still had make some changes for support with merge handles https://github.com/apache/hudi/pull/18539#issuecomment-4427122042 -- 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]
Re: [I] Flink inference of Hudi unstructured logical types (Variant, Blob, Vector) from Parquet [hudi]
danny0405 commented on issue #18711: URL: https://github.com/apache/hudi/issues/18711#issuecomment-4426930513 Hi, @kbuci , in this PR: https://github.com/apache/hudi/pull/18717, I migrated the current ITs to run with Flink 2.1, and did very small changes on `ParquetSplitReaderUtil`, and the reader already works, there is no need to pass around explciit schema on reader side. `ParquetSplitReaderUtil` read it physically as row and Flink runtime will translate it into `Variant` data type. -- 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]
Re: [I] Flink inference of Hudi unstructured logical types (Variant, Blob, Vector) from Parquet [hudi]
kbuci commented on issue #18711: URL: https://github.com/apache/hudi/issues/18711#issuecomment-4423834271 @cshuo > The Hudi read path should use the writer/table HoodieSchema to read the file and then project/convert to the requested schema. Thanks for the clarification! - when I first encountered this issue my thought process was to just check if we can make sure generally all HUDI Flink read/write code paths "infer" with the HoodieSchema. But I think we should instead follow the model you highlighted - this schema inference logic should only need to happen at "read" time. Since thats anyway the current precedent in Spark and Flink. I'm updating my existing Flink variant PR to go with this approach (which is still similar to approach A in essence). https://github.com/apache/hudi/pull/18539 Since once this is wired-up, Blob and Vector implementations should be easier to understand/add -- 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]
Re: [I] Flink inference of Hudi unstructured logical types (Variant, Blob, Vector) from Parquet [hudi]
cshuo commented on issue #18711: URL: https://github.com/apache/hudi/issues/18711#issuecomment-4419033901 > Spark's read path already has HoodieSchema available at the point where Parquet → Spark type conversion happens. Small clarification on the Spark comparison, `HoodieSparkParquetReader#getSchema()` still reads the Parquet `MessageType`, converts it to Spark `StructType`, and then converts that to `HoodieSchema`; it is not using `HoodieSchema` to infer Hudi logical types there. The important part is the actual record read path. Spark does not rely on `getSchema()` / Parquet physical inference to recover VECTOR semantics. In `getUnsafeRowIterator(requestedSchema)`, it uses the requested/writer `HoodieSchema` to detect vector columns, rewrites those columns to `BinaryType` for the physical Parquet read, and then converts the binary value back to the logical vector array after reading. So for Flink, I think the same principle should apply: avoid making `Parquet MessageType -> RowType` responsible for recovering Hudi-specific logical types. The Hudi read path should use the writer/table `HoodieSchema` to read the file and then project/convert to the requested schema. `FlinkRowDataReaderContext#getFileRecordIterator` already follows this model by receiving `dataSchema` and constructing the `RowType` from it before reading. Given that, I would prefer Option A. The rationale is that the real Hudi read semantics should be schema-driven, and Parquet schema conversion should remain a generic/best-effort fallback rather than the authority for Hudi logical types. -- 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]
Re: [I] Flink inference of Hudi unstructured logical types (Variant, Blob, Vector) from Parquet [hudi]
kbuci commented on issue #18711: URL: https://github.com/apache/hudi/issues/18711#issuecomment-4411765009 Yes, I'd prefer that as well - passing around the Hoodie schema as the "authoritative source" to infer HUDI logical types, since thats more understandable/clean and any seems to be the precedent in HUDI spark. My only concern was figuring out if that breaks any public APIs, but I can assess that as I create the PR, and if so we can make sure to land any such PR in 1.3+ (and not in 1.2). I'm still famializing myself with other table formats, but based on a very rough/initial search it seems other table formats might also be leaning towards A approach in practice? ## Iceberg - **Iceberg Spec (v3, includes Variant type definition)**: https://iceberg.apache.org/spec - **Iceberg Schemas doc (field IDs, type system)**: https://iceberg.apache.org/docs/latest/schemas - **PR: Add variant type support to ParquetTypeVisitor**: https://github.com/apache/iceberg/pull/14588 - **PR: Implement Variant Parquet readers**: https://github.com/apache/iceberg/pull/12139 - **PR: Spec — add variant type**: https://github.com/apache/iceberg/pull/10831 - **Snowflake blog: Iceberg v3 Variant Type**: https://www.snowflake.com/en/engineering-blog/apache-iceberg-v3-variant-type/ ## Delta Lake - **Delta Variant Type RFC**: https://github.com/delta-io/delta/blob/master/protocol_rfcs/accepted/variant-type.md - **Delta Protocol (v4.2.0, schema in transaction log)**: https://github.com/delta-io/delta/blob/v4.2.0/PROTOCOL.md - **PR: Add VariantType support in Spark schema conversion**: https://github.com/delta-io/delta/pull/6164 - **PR: Kernel-level variant schema deserialization**: https://github.com/delta-io/delta/pull/3464 - **Delta Variant Shredding RFC**: https://github.com/delta-io/delta/blob/master/protocol_rfcs/variant-shredding.md ## Paimon - **PIP-40: Introduce a new Vector data type**: https://cwiki.apache.org/confluence/display/PAIMON/PIP-40%3A+Introduce+a+new+Vector+data+type - **Issue: Introduce VecType**: https://github.com/apache/paimon/issues/7011 - **PR: Add Flink support for VectorType**: https://github.com/apache/paimon/pull/7238 - **Paimon FileFormat spec (Parquet type mapping)**: https://paimon.apache.org/docs/1.4/concepts/spec/fileformat/ - **Paimon ParquetSchemaConverter API**: https://paimon.apache.org/docs/0.9/api/java/org/apache/paimon/format/parquet/ParquetSchemaConverter.html -- 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]
Re: [I] Flink inference of Hudi unstructured logical types (Variant, Blob, Vector) from Parquet [hudi]
danny0405 commented on issue #18711: URL: https://github.com/apache/hudi/issues/18711#issuecomment-4411415157 Thanks for rasing this @kbuci , it deserves a discussion. Technically I prefer A if passing around the schema can solve the issue without add footer metadata keys, it looks like a chaos that these 3 data types have differnt handling of the storage of the new metadata keys, wondering if we could have a good way to unify them, will investigate a little bit like how Iceberg handles this. -- 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]
[I] Flink inference of Hudi unstructured logical types (Variant, Blob, Vector) from Parquet [hudi]
kbuci opened a new issue, #18711:
URL: https://github.com/apache/hudi/issues/18711
## Problem Statement
Flink's type conversion process currently cannot do a 1:1 inference of
Hudi's unstructured logical types (**Variant**, **Blob**, **Vector**) to the
appropriate Flink data type from Parquet schema alone. The Parquet physical
layout for these types is ambiguous without additional context:
| Hudi Type | Parquet Physical Layout | Ambiguity |
|---||---|
| **Variant** | `GROUP { required binary metadata; required binary value; }`
| Indistinguishable from a user struct `ROW` |
| **Vector** | `FIXED_LEN_BYTE_ARRAY(N)` | Indistinguishable from any
fixed-length binary field |
| **Blob** | Nested group with canonical fields (`type`, `data`,
`reference`) | Indistinguishable from a user struct with those field names |
This is because Hudi compiles against **parquet-java 1.13.1** by default
([pom.xml L118](https://github.com/apache/hudi/blob/master/pom.xml#L118)),
which predates `VariantLogicalTypeAnnotation` (introduced in parquet-java
1.15.2+). There is no standard Parquet `LogicalTypeAnnotation` for Blob or
Vector at all.
### How does the Spark path handle this today?
On the **Spark side**, Hudi uses other mechanisms rather than Parquet schema
annotations:
- **Vector**: Writes `hoodie.vector.columns` key-value metadata in the
Parquet file footer
([HoodieRowParquetWriteSupport.java](https://github.com/apache/hudi/blob/master/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java)),
plus `hudi_type` metadata on Spark `StructField`. Detection on read uses the
requested `HoodieSchema` or Spark field metadata — never the Parquet column
schema alone
([VectorConversionUtils.java](https://github.com/apache/hudi/blob/master/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/VectorConversionUtils.java)).
- **Blob**: Uses Avro `logicalType: blob` and Spark `hudi_type: BLOB`
metadata. No dedicated Parquet footer key.
- **Variant**: Currently no footer metadata. The Parquet
`VariantLogicalTypeAnnotation` is only available at runtime if parquet-java ≥
1.15.2 is on the classpath. The existing `hasVariantAnnotation()` check uses
**class-name string matching** to avoid compile-time dependency
([ParquetSchemaConverter.java
L430](https://github.com/apache/hudi/blob/master/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/parquet/ParquetSchemaConverter.java#L430)).
### Why is this harder for Flink than Spark?
Spark's read path already has `HoodieSchema` available at the point where
Parquet → Spark type conversion happens. Flink's
`ParquetSchemaConverter.convertToRowType(MessageType)` was originally a
**standalone** conversion that only looked at the Parquet `MessageType` — no
HUDI avro schema information.
---
## Proposed Options
### Option A: Require `HoodieSchema` in all Flink Parquet-to-type conversions
**Approach**: Make `HoodieSchema` a **required** parameter (not
optional/nullable) in
[`ParquetSchemaConverter.convertToRowType()`](https://github.com/apache/hudi/blob/master/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/parquet/ParquetSchemaConverter.java#L77)
and thread it through all Flink read entry points
([`HoodieRowDataParquetReader`](https://github.com/apache/hudi/blob/master/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieRowDataParquetReader.java),
[`FlinkRowDataReaderContext`](https://github.com/apache/hudi/blob/master/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FlinkRowDataReaderContext.java),
`ClusteringOperator`, `CopyOnWriteInputFormat`, `MergeOnReadInputFormat`,
etc.).
**Pros**:
- Works immediately with current parquet-java 1.13.1
- Handles all three types (Variant, Blob, Vector) uniformly
- Can read unstructured data written by Spark (current HUDI builds) today
- Consistent with Spark's approach (Spark always has `HoodieSchema`
available at conversion time)
**Cons**:
- Touches many Flink read path entry points — potentially breaking internal
APIs
- Every new Flink reader/operator that does Parquet→type conversion must
remember to supply the schema
- Does this break any public connector APIs? (Need to verify —
`HoodieTableFactory` / `DynamicTableSource` wiring)
**Open question**: Does this actually break any *public-facing* API (`CREATE
TABLE` DDL options, `FlinkOptions`, etc.), or only internal plumbing?
---
### Option B: Parquet file footer metadata (like `hoodie.vector.columns`)
**Approach**: On the write side (both Spark and Flink), emit footer
key-value metadata listing which columns are Variant/Blob (e.g.,
`hoodie.variant.columns`, `hoodie.blob.columns`). On the Flink read
