hudi-agent commented on code in PR #18723:
URL: https://github.com/apache/hudi/pull/18723#discussion_r3336131661
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/DataTypeUtils.java:
##########
@@ -120,6 +124,48 @@ public static int[] projectOrdinals(RowType rowType,
RowType producedRowType) {
return
producedRowType.getFieldNames().stream().mapToInt(fieldNames::indexOf).toArray();
}
+ /**
+ * Creates a hoodie schema from a Flink row type with logical metadata from
the table schema.
+ *
+ * <p>When a field is a hoodie specific logical type in {@code tableSchema},
this method
+ * reuses the table schema field to preserve logical metadata that cannot be
recovered from Flink
+ * {@link RowType}, for example VECTOR element type and dimension. Other
fields are taken from the
+ * schema converted from {@code rowType}, so the returned schema follows the
row type's field order
+ * while retaining hoodie-specific logical metadata where needed.
+ *
+ * @param rowType Flink row type to convert
+ * @param tableSchema source table schema with hoodie logical type metadata
+ * @return hoodie schema matching the row type field order
+ */
+ public static HoodieSchema toHoodieSchemaWithLogicalMetadata(RowType
rowType, HoodieSchema tableSchema) {
+ HoodieSchema convertedSchema =
HoodieSchemaConverter.convertToSchema(rowType);
+ List<HoodieSchemaField> schemaFields = new
ArrayList<>(rowType.getFieldCount());
+
+ for (String fieldName : rowType.getFieldNames()) {
+ HoodieSchemaField tableField =
tableSchema.getField(fieldName).orElse(null);
+ HoodieSchemaField field = tableField != null &&
useTableSchemaField(tableField)
+ ? tableField : convertedSchema.getField(fieldName).get();
+ schemaFields.add(HoodieSchemaUtils.createNewSchemaField(field));
+ }
+
+ return HoodieSchema.createRecord(
+ tableSchema.getName(),
+ tableSchema.getNamespace().orElse(null),
+ tableSchema.getDoc().orElse(null),
+ schemaFields);
+ }
+
+ /**
+ * Returns whether the converted schema should reuse the field from the
table schema.
+ *
+ * <p>Only types whose logical metadata cannot be fully reconstructed from
Flink
+ * {@link RowType} are reused from the table schema.
+ */
+ private static boolean useTableSchemaField(HoodieSchemaField field) {
+ HoodieSchemaType fieldType = field.schema().getNonNullType().getType();
Review Comment:
🤖 nit: `useTableSchemaField` doesn't read as a predicate — it's not
immediately clear what condition it's testing. Could you consider a name like
`hasHoodieOnlyLogicalType` or `requiresLogicalMetadataFromTableSchema`? The
existing Javadoc already describes the intent well, it just needs the name to
match.
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/util/HoodieVectorUtils.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.util;
+
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaField;
+import org.apache.hudi.common.schema.HoodieSchemaType;
+
+import java.nio.ByteBuffer;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Utilities for decoding Hudi VECTOR fixed-bytes payloads.
+ */
+public final class HoodieVectorUtils {
+
+ private HoodieVectorUtils() {
+ }
+
+ /**
+ * Detects VECTOR columns in a HoodieSchema record and returns a map of
field ordinal
+ * to the corresponding {@link HoodieSchema.Vector} schema.
+ *
+ * @param schema a HoodieSchema of type RECORD (or null)
+ * @return map from field index to Vector schema; empty map if schema is
null or has no vectors
+ */
+ public static Map<Integer, HoodieSchema.Vector>
detectVectorColumns(HoodieSchema schema) {
+ Map<Integer, HoodieSchema.Vector> vectorColumnInfo = new LinkedHashMap<>();
+ if (schema == null) {
+ return vectorColumnInfo;
+ }
+ List<HoodieSchemaField> fields = schema.getFields();
+ for (int i = 0; i < fields.size(); i++) {
+ HoodieSchema fieldSchema = fields.get(i).schema().getNonNullType();
+ if (fieldSchema.getType() == HoodieSchemaType.VECTOR) {
+ vectorColumnInfo.put(i, (HoodieSchema.Vector) fieldSchema);
+ }
+ }
+ return vectorColumnInfo;
+ }
+
+ /**
+ * Converts binary bytes from a FIXED_LEN_BYTE_ARRAY Parquet column back to
a typed array.
+ *
+ * @param bytes raw bytes read from Parquet
+ * @param vectorSchema vector schema
+ * @return an ArrayData containing the decoded float[], double[], or byte[]
array
+ * @throws IllegalArgumentException if byte array length doesn't match
expected size
+ */
Review Comment:
🤖 nit: `@return an ArrayData containing...` — `ArrayData` is a Flink/Spark
abstraction but this class lives in `hudi-common`. The method actually returns
a plain Java primitive array (`float[]`, `double[]`, or `byte[]`) as `Object`.
Could the `@return` be updated to something like "a primitive array (`float[]`,
`double[]`, or `byte[]`) depending on the element type"?
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/AvroToRowDataConverters.java:
##########
@@ -78,16 +82,55 @@ public interface AvroToRowDataConverter extends
Serializable {
//
-------------------------------------------------------------------------------------
// Runtime Converters
//
-------------------------------------------------------------------------------------
+
+ /**
+ * Creates a row converter from the Flink row type using UTC timezone
conversion.
+ *
+ * <p>Note that this RowType-only path cannot recover Hoodie-specific
logical type metadata
+ * that is not represented in Flink's {@link RowType}.
+ */
public static AvroToRowDataConverter createRowConverter(RowType rowType) {
return createRowConverter(rowType, true);
}
+ /**
+ * Creates a row converter from the Flink row type.
+ *
+ * <p>Note that this RowType-only path cannot recover Hoodie-specific
logical type metadata
+ * that is not represented in Flink's {@link RowType}.
+ */
public static AvroToRowDataConverter createRowConverter(RowType rowType,
boolean utcTimezone) {
- final AvroToRowDataConverter[] fieldConverters =
- rowType.getFields().stream()
- .map(RowType.RowField::getType)
- .map(type -> AvroToRowDataConverters.createNullableConverter(type,
utcTimezone))
- .toArray(AvroToRowDataConverter[]::new);
+ return createRowConverter(HoodieSchemaConverter.convertToSchema(rowType),
rowType, utcTimezone);
+ }
+
+ /**
+ * Creates a row converter using only the Flink row type.
Review Comment:
🤖 nit: the Javadoc here says "Creates a row converter using only the Flink
row type" but the method signature is `createRowConverter(HoodieSchema
hoodieSchema)` — it actually has the Hoodie schema and *can* recover VECTOR
metadata. The paragraph below that says "use the 3-arg overload when a
HoodieSchema is available" is also backwards. Could you update the description
to reflect that this overload derives the RowType from the HoodieSchema
internally, e.g. "Creates a row converter from a Hoodie schema, deriving the
target RowType internally with UTC timezone"?
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
