hudi-agent commented on code in PR #19842:
URL: https://github.com/apache/hudi/pull/19842#discussion_r3947636831
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/VectorColumnParser.java:
##########
@@ -69,8 +68,7 @@ public static Map<String, Integer> parse(String
vectorColumns) {
"Invalid VECTOR column descriptor '" + entry + "'. Expected
format: columnName[:dimension].");
}
String columnName = parts[0].trim();
- String normalizedColumnName = columnName.toLowerCase(Locale.ROOT);
- if (parsed.containsKey(normalizedColumnName)) {
+ if (parsed.containsKey(columnName)) {
Review Comment:
🤖 This goes further than preserving file-column casing: it makes the
user-facing `hoodie.vector.columns` option case-sensitive on the write path too
(`HoodieTableFactory`, `HoodieHiveCatalog.initTableIfNotExists`,
`BulkInsertWriterHelper`, `ClusteringOperator`). With `HoodieHiveCatalog`,
`getTable` falls back to the HMS column list (lower-cased by the metastore)
until the first commit exists, so a table declared with `Embedding` +
`'hoodie.vector.columns'='Embedding:4'` could now fail with "VECTOR column
'Embedding' does not exist" on its first INSERT, where it previously matched.
Is that intended? @danny0405 you asked for case preservation on the file side —
is case-sensitive option matching the behavior you want as well?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieRowDataLanceReader.java:
##########
@@ -204,7 +211,53 @@ public ClosableIterator<RowData>
getRowDataIterator(DataType dataType, HoodieSch
@Override
public HoodieSchema getSchema() {
RowType rowType = HoodieFlinkLanceArrowUtils.toRowType(arrowSchema);
- return HoodieSchemaConverter.convertToSchema(rowType);
+ Map<String, String> metadata = arrowSchema.getCustomMetadata();
+ Set<String> vectorColumnNames = HoodieSchema.parseVectorColumnNames(
+ metadata == null ? null :
metadata.get(HoodieSchema.VECTOR_COLUMNS_METADATA_KEY));
+ if (vectorColumnNames.isEmpty()) {
+ return HoodieSchemaConverter.convertToSchema(rowType);
+ }
+ String vectorColumns = vectorColumnNames.stream()
+ .map(name -> name + ":" +
vectorSchemaFromArrow(getTopLevelField(name)).getDimension())
+ .collect(Collectors.joining(","));
+ return HoodieSchemaConverter.convertToSchema(rowType, "record",
vectorColumns);
+ }
+
+ private void validateRequestedVectors(HoodieSchema requestedSchema) {
+ for (HoodieSchemaField field :
requestedSchema.getNonNullType().getFields()) {
+ HoodieSchema fieldSchema = field.schema().getNonNullType();
+ if (fieldSchema.getType() != HoodieSchemaType.VECTOR) {
+ continue;
+ }
+ HoodieSchema.Vector expected = (HoodieSchema.Vector) fieldSchema;
+ HoodieSchema.Vector actual =
vectorSchemaFromArrow(getTopLevelField(field.name()));
+ if (actual.getDimension() != expected.getDimension()
+ || actual.getVectorElementType() != expected.getVectorElementType())
{
+ throw new HoodieValidationException(
+ "Incompatible Lance VECTOR encoding for column '" + field.name()
+ + "': requested " + expected.toTypeDescriptor()
+ + " but file contains " + actual.toTypeDescriptor());
+ }
+ }
+ }
+
+ private Field getTopLevelField(String name) {
+ return arrowSchema.getFields().stream()
+ .filter(field -> field.getName().equals(name))
+ .findFirst()
+ .orElseThrow(() -> new HoodieValidationException(
+ "Missing Lance column in file schema: " + name));
+ }
+
+ private static HoodieSchema.Vector vectorSchemaFromArrow(Field field) {
+ ArrowType.FixedSizeList listType = (ArrowType.FixedSizeList)
field.getType();
Review Comment:
🤖 These two casts are unchecked, so a column that isn't
`FixedSizeList<Float32|Float64>` fails with a raw `ClassCastException` rather
than the `HoodieValidationException` the surrounding code promises. That's
reachable today: Flink Lance files written on master between #18877 and #19831
encoded VECTOR-typed columns as a variable-length `List` (no footer key), and
with the COW path now requesting VECTOR from the table schema,
`validateRequestedVectors` will hit that cast. Could this check `instanceof`
and throw a descriptive validation error instead?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieRowDataLanceReader.java:
##########
@@ -204,7 +211,53 @@ public ClosableIterator<RowData>
getRowDataIterator(DataType dataType, HoodieSch
@Override
public HoodieSchema getSchema() {
RowType rowType = HoodieFlinkLanceArrowUtils.toRowType(arrowSchema);
- return HoodieSchemaConverter.convertToSchema(rowType);
+ Map<String, String> metadata = arrowSchema.getCustomMetadata();
+ Set<String> vectorColumnNames = HoodieSchema.parseVectorColumnNames(
+ metadata == null ? null :
metadata.get(HoodieSchema.VECTOR_COLUMNS_METADATA_KEY));
+ if (vectorColumnNames.isEmpty()) {
+ return HoodieSchemaConverter.convertToSchema(rowType);
+ }
+ String vectorColumns = vectorColumnNames.stream()
Review Comment:
🤖 nit: building a `name:dim,name:dim` string here only for
`VectorColumnParser` to split it back apart feels roundabout — have you
considered adding a `convertToSchema(RowType, String, Map<String, Integer>)`
overload and passing the map directly?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieRowDataLanceReader.java:
##########
@@ -204,7 +211,53 @@ public ClosableIterator<RowData>
getRowDataIterator(DataType dataType, HoodieSch
@Override
public HoodieSchema getSchema() {
RowType rowType = HoodieFlinkLanceArrowUtils.toRowType(arrowSchema);
- return HoodieSchemaConverter.convertToSchema(rowType);
+ Map<String, String> metadata = arrowSchema.getCustomMetadata();
+ Set<String> vectorColumnNames = HoodieSchema.parseVectorColumnNames(
+ metadata == null ? null :
metadata.get(HoodieSchema.VECTOR_COLUMNS_METADATA_KEY));
+ if (vectorColumnNames.isEmpty()) {
+ return HoodieSchemaConverter.convertToSchema(rowType);
+ }
+ String vectorColumns = vectorColumnNames.stream()
+ .map(name -> name + ":" +
vectorSchemaFromArrow(getTopLevelField(name)).getDimension())
+ .collect(Collectors.joining(","));
+ return HoodieSchemaConverter.convertToSchema(rowType, "record",
vectorColumns);
+ }
+
+ private void validateRequestedVectors(HoodieSchema requestedSchema) {
+ for (HoodieSchemaField field :
requestedSchema.getNonNullType().getFields()) {
+ HoodieSchema fieldSchema = field.schema().getNonNullType();
+ if (fieldSchema.getType() != HoodieSchemaType.VECTOR) {
+ continue;
+ }
+ HoodieSchema.Vector expected = (HoodieSchema.Vector) fieldSchema;
+ HoodieSchema.Vector actual =
vectorSchemaFromArrow(getTopLevelField(field.name()));
+ if (actual.getDimension() != expected.getDimension()
+ || actual.getVectorElementType() != expected.getVectorElementType())
{
+ throw new HoodieValidationException(
+ "Incompatible Lance VECTOR encoding for column '" + field.name()
+ + "': requested " + expected.toTypeDescriptor()
+ + " but file contains " + actual.toTypeDescriptor());
+ }
+ }
+ }
+
+ private Field getTopLevelField(String name) {
+ return arrowSchema.getFields().stream()
+ .filter(field -> field.getName().equals(name))
+ .findFirst()
+ .orElseThrow(() -> new HoodieValidationException(
+ "Missing Lance column in file schema: " + name));
+ }
+
+ private static HoodieSchema.Vector vectorSchemaFromArrow(Field field) {
+ ArrowType.FixedSizeList listType = (ArrowType.FixedSizeList)
field.getType();
+ ArrowType.FloatingPoint elementType =
+ (ArrowType.FloatingPoint) field.getChildren().get(0).getType();
+ HoodieSchema.Vector.VectorElementType vectorElementType =
+ elementType.getPrecision() == FloatingPointPrecision.SINGLE
Review Comment:
🤖 nit: this silently maps anything that isn't SINGLE (e.g. HALF, or a
non-FloatingPoint child after the cast) to DOUBLE; it might be worth an
explicit `checkArgument`/`unsupported` branch so a malformed file fails with a
clear message instead of a ClassCastException or a wrong 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]
