hudi-agent commented on code in PR #18723:
URL: https://github.com/apache/hudi/pull/18723#discussion_r3246427181
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/AvroToRowDataConverters.java:
##########
@@ -143,19 +173,34 @@ public static AvroToRowDataConverter
createConverter(LogicalType type, boolean u
return createTimestampConverter(((TimestampType) type).getPrecision(),
utcTimezone);
case CHAR:
case VARCHAR:
- return avroObject -> avroObject instanceof Utf8 ?
StringData.fromBytes(((Utf8) avroObject).getBytes()) :
StringData.fromString(avroObject.toString());
+ return avroObject -> avroObject instanceof Utf8
+ ? StringData.fromBytes(((Utf8) avroObject).getBytes())
+ : StringData.fromString(avroObject.toString());
case BINARY:
case VARBINARY:
return AvroToRowDataConverters::convertToBytes;
case DECIMAL:
return createDecimalConverter((DecimalType) type);
case ARRAY:
- return createArrayConverter((ArrayType) type, utcTimezone);
+ if (nonNullSchema.getType() == HoodieSchemaType.VECTOR) {
+ HoodieSchema.Vector vectorSchema = (HoodieSchema.Vector)
nonNullSchema;
+ VectorConversionUtils.validateVectorLogicalType(vectorSchema, type);
+ return createVectorConverter(vectorSchema);
+ }
+ return createArrayConverter(nonNullSchema.getElementType(),
(ArrayType) type, utcTimezone);
case ROW:
- return createRowConverter((RowType) type, utcTimezone);
+ if (nonNullSchema.getType() == HoodieSchemaType.UNION) {
+ // getNonNullType() unwraps simple nullable unions only. Complex
unions can still reach
+ // here when their Flink representation is a ROW, for example fields
inside
+ // ColumnStatsSchemas.METADATA_SCHEMA. In that case the RowType
already captures the
+ // target Flink shape, so use the first union branch only as the
positional Hoodie schema
+ // template for building nested field converters.
Review Comment:
🤖 Could you confirm that for every complex union reaching this branch (e.g.
fields inside `ColumnStatsSchemas.METADATA_SCHEMA`), the first union branch
always has the same positional field shape as the target Flink `RowType`?
Picking `getTypes().get(0)` and then indexing into its fields positionally will
silently produce wrong converters if a future union has heterogeneous record
members or the first branch is not a record.
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/DataTypeUtils.java:
##########
@@ -120,6 +123,36 @@ public static int[] projectOrdinals(RowType rowType,
RowType producedRowType) {
return
producedRowType.getFieldNames().stream().mapToInt(fieldNames::indexOf).toArray();
}
+ /**
+ * Creates the hoodie required schema for a projected Flink row type.
+ *
+ * <p>When a requested field exists in {@code tableSchema}, this method
reuses the table schema
+ * field to preserve hoodie-specific logical metadata that cannot be
recovered from Flink
+ * {@link RowType}, for example VECTOR element type and dimension. When a
requested field does
+ * not exist in {@code tableSchema}, the field is taken from the schema
converted from
+ * {@code requiredRowType}, so readers can still keep missing required
columns in the requested
+ * schema for later schema-evolution/default-value handling.
+ *
+ * @param tableSchema source table schema with hoodie logical type
metadata
+ * @param requiredRowType projected Flink row type requested by the query
+ * @return required hoodie schema matching the projected field order
+ */
+ public static HoodieSchema createRequiredSchema(HoodieSchema tableSchema,
RowType requiredRowType) {
+ HoodieSchema fallbackRequiredSchema =
HoodieSchemaConverter.convertToSchema(requiredRowType);
+ List<HoodieSchemaField> requiredFields = new
ArrayList<>(requiredRowType.getFieldCount());
+
+ for (String fieldName : requiredRowType.getFieldNames()) {
+ HoodieSchemaField field =
tableSchema.getField(fieldName).orElse(fallbackRequiredSchema.getField(fieldName).get());
Review Comment:
🤖 Minor: the chained
`orElse(fallbackRequiredSchema.getField(fieldName).get())` will NPE with a
confusing message if `convertToSchema(requiredRowType).getField(fieldName)`
ever returns empty. Today this should always succeed since
`fallbackRequiredSchema` is built directly from `requiredRowType`, but
`orElseThrow(() -> new HoodieException("Field " + fieldName + " not found in
either tableSchema or requiredRowType"))` would fail more clearly if that
assumption breaks.
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink1.17.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -365,7 +365,15 @@ private static ColumnReader createColumnReader(
case VARCHAR:
case BINARY:
case VARBINARY:
- return new BytesColumnReader(descriptor, pageReader);
+ switch (descriptor.getPrimitiveType().getPrimitiveTypeName()) {
+ case BINARY:
+ return new BytesColumnReader(descriptor, pageReader);
+ case FIXED_LEN_BYTE_ARRAY:
+ return new FixedLenBytesColumnReader(
+ descriptor, pageReader);
+ default:
+ throw new AssertionError();
Review Comment:
🤖 nit: `throw new AssertionError()` with no message makes the failure hard
to diagnose. The 1.18.x copy of this file includes a descriptive string
("Unexpected physical type for BYTES: " + typeName) — could you add the same
message here and in the other flink version copies (1.19.x, 1.20.x, 2.0.x,
2.1.x)?
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/DataTypeUtils.java:
##########
@@ -120,6 +123,36 @@ public static int[] projectOrdinals(RowType rowType,
RowType producedRowType) {
return
producedRowType.getFieldNames().stream().mapToInt(fieldNames::indexOf).toArray();
}
+ /**
+ * Creates the hoodie required schema for a projected Flink row type.
+ *
+ * <p>When a requested field exists in {@code tableSchema}, this method
reuses the table schema
+ * field to preserve hoodie-specific logical metadata that cannot be
recovered from Flink
+ * {@link RowType}, for example VECTOR element type and dimension. When a
requested field does
+ * not exist in {@code tableSchema}, the field is taken from the schema
converted from
+ * {@code requiredRowType}, so readers can still keep missing required
columns in the requested
+ * schema for later schema-evolution/default-value handling.
+ *
+ * @param tableSchema source table schema with hoodie logical type
metadata
+ * @param requiredRowType projected Flink row type requested by the query
+ * @return required hoodie schema matching the projected field order
+ */
+ public static HoodieSchema createRequiredSchema(HoodieSchema tableSchema,
RowType requiredRowType) {
+ HoodieSchema fallbackRequiredSchema =
HoodieSchemaConverter.convertToSchema(requiredRowType);
+ List<HoodieSchemaField> requiredFields = new
ArrayList<>(requiredRowType.getFieldCount());
+
+ for (String fieldName : requiredRowType.getFieldNames()) {
Review Comment:
🤖 nit: `orElse(fallbackRequiredSchema.getField(fieldName).get())` eagerly
evaluates `.get()` on every iteration even when `tableSchema` already has the
field, and a bare `.get()` hides the invariant being asserted. Consider
`orElseGet(() -> fallbackRequiredSchema.getField(fieldName).orElseThrow())` to
make the laziness and the expectation explicit.
<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
Review Comment:
🤖 nit: the `@return` says "an ArrayData" but `ArrayData` is a Spark type —
this class lives in `hudi-common` with no Spark dependency, and the method
actually returns `float[]`, `double[]`, or `byte[]`. Could you update the
`@return` (here and on line 77) to something like "a `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 +81,27 @@ public interface AvroToRowDataConverter extends
Serializable {
//
-------------------------------------------------------------------------------------
// Runtime Converters
//
-------------------------------------------------------------------------------------
- public static AvroToRowDataConverter createRowConverter(RowType rowType) {
- return createRowConverter(rowType, true);
+ /**
+ * Creates a row converter using only the Flink row type.
+ *
+ * <p>This converter cannot recover Hoodie-specific logical type metadata
from {@link RowType}.
+ * Use {@link #createRowConverter(HoodieSchema, RowType, boolean)} when a
Hoodie schema is
+ * available, especially for VECTOR columns.
+ */
+ public static AvroToRowDataConverter createRowConverter(HoodieSchema
hoodieSchema) {
+ return createRowConverter(hoodieSchema, (RowType)
HoodieSchemaConverter.convertToDataType(hoodieSchema).getLogicalType(), true);
}
- 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);
+ /**
+ * Creates a row converter using both Hoodie schema metadata and the target
Flink row type.
+ */
+ public static AvroToRowDataConverter createRowConverter(HoodieSchema schema,
RowType rowType, boolean utcTimezone) {
+ HoodieSchema recordSchema = schema.getNonNullType();
+ final List<HoodieSchemaField> fields = recordSchema.getFields();
+ final AvroToRowDataConverter[] fieldConverters = new
AvroToRowDataConverter[rowType.getFieldCount()];
+ for (int i = 0; i < fieldConverters.length; i++) {
+ fieldConverters[i] = createNullableConverter(fields.get(i).schema(),
rowType.getTypeAt(i), utcTimezone);
Review Comment:
🤖 The new `createRowConverter(HoodieSchema, RowType, boolean)` implicitly
requires that `schema.getNonNullType().getFields()` and `rowType.getFields()`
are positionally aligned and have the same count — otherwise
`fields.get(i).schema()` either throws IndexOutOfBoundsException or silently
produces a wrong converter. All current callers preserve this, but it'd be
worth a one-line javadoc note or `checkArgument(fields.size() >=
rowType.getFieldCount())` so future callers don't have to discover this the
hard way.
<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]
