hudi-agent commented on code in PR #18723:
URL: https://github.com/apache/hudi/pull/18723#discussion_r3301157291
##########
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: the `default` branch here (and in the 1.19.x / 1.20.x / 2.0.x / 2.1.x
copies of this file) throws `new AssertionError()` with no message, while the
1.18.x version includes a helpful diagnostic string. Could you add a message
like `"Unexpected physical type for BYTES: " +
descriptor.getPrimitiveType().getPrimitiveTypeName()` to match the 1.18.x
pattern and make the failure actionable in a stack trace?
<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:
🤖 nit: calling `.get()` directly on the Optional from
`fallbackRequiredSchema.getField(fieldName)` will throw a bare
`NoSuchElementException` with no context if the field is somehow absent from
both schemas. Could you replace it with `.orElseThrow(() -> new
IllegalStateException("Field not found in fallback schema: " + fieldName))` to
make any future failures easier to diagnose?
<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
+ */
+ public static Object decodeVectorBytes(byte[] bytes, HoodieSchema.Vector
vectorSchema) {
Review Comment:
🤖 nit: the `@return` Javadoc on both `decodeVectorBytes` overloads says "an
ArrayData containing the decoded..." but the return type is `Object` (a raw
`float[]`, `double[]`, or `byte[]`). `ArrayData` is a Flink type that doesn't
live in `hudi-common`, so this is a bit misleading — could you update the
`@return` to say 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>
--
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]
