This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new d9e6f42045cd feat(flink): support vector columns in Lance reader
(#19842)
d9e6f42045cd is described below
commit d9e6f42045cdf69f0d3f6f9627125338cbfeee39
Author: Shuo Cheng <[email protected]>
AuthorDate: Tue Sep 8 12:18:47 2026 +0800
feat(flink): support vector columns in Lance reader (#19842)
* feat(flink): support vector columns in Lance reader
---
.../row/lance/HoodieFlinkLanceArrowUtils.java | 22 +-
.../apache/hudi/util/HoodieSchemaConverter.java | 18 +-
.../org/apache/hudi/util/VectorColumnParser.java | 14 +-
.../apache/hudi/util/VectorConversionUtils.java | 9 +-
.../hudi/util/TestHoodieSchemaConverter.java | 16 ++
.../hudi/util/TestVectorConversionUtils.java | 14 +-
.../org/apache/hudi/table/format/FormatUtils.java | 15 +-
.../table/format/HoodieRowDataLanceReader.java | 78 ++++++-
.../table/format/cow/CopyOnWriteInputFormat.java | 20 +-
.../apache/hudi/table/ITTestVectorDataSource.java | 60 ++++++
.../table/format/TestHoodieRowDataLanceReader.java | 229 ++++++++++++++++++++-
.../format/cow/TestCopyOnWriteInputFormat.java | 63 ++++++
12 files changed, 515 insertions(+), 43 deletions(-)
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/lance/HoodieFlinkLanceArrowUtils.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/lance/HoodieFlinkLanceArrowUtils.java
index 9806f5e9ad1e..896ce10af1bb 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/lance/HoodieFlinkLanceArrowUtils.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/lance/HoodieFlinkLanceArrowUtils.java
@@ -40,6 +40,8 @@ import org.apache.arrow.vector.TinyIntVector;
import org.apache.arrow.vector.ValueVector;
import org.apache.arrow.vector.VarBinaryVector;
import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.complex.BaseListVector;
+import org.apache.arrow.vector.complex.FixedSizeListVector;
import org.apache.arrow.vector.complex.ListVector;
import org.apache.arrow.vector.complex.StructVector;
import org.apache.arrow.vector.types.DateUnit;
@@ -174,7 +176,7 @@ public final class HoodieFlinkLanceArrowUtils {
case ROW:
return readRow((RowType) type, (StructVector) vector, rowId);
case ARRAY:
- return readArray((ArrayType) type, (ListVector) vector, rowId);
+ return readArray((ArrayType) type, vector, rowId);
default:
throw unsupported(type);
}
@@ -219,6 +221,9 @@ public final class HoodieFlinkLanceArrowUtils {
private static void validateLanceVector(String fieldName,
HoodieSchema.Vector vectorSchema) {
HoodieSchema.Vector.VectorElementType elementType =
vectorSchema.getVectorElementType();
+ // Keep the on-disk encoding aligned with Spark Lance writes. The current
lance-spark
+ // VectorUtils.shouldBeFixedSizeList recognizes only Array<Float> and
Array<Double>;
+ // an INT8 vector would otherwise be emitted as a variable-size List
instead of FixedSizeList.
if (elementType != HoodieSchema.Vector.VectorElementType.FLOAT
&& elementType != HoodieSchema.Vector.VectorElementType.DOUBLE) {
throw new HoodieNotSupportedException(
@@ -319,9 +324,9 @@ public final class HoodieFlinkLanceArrowUtils {
fields.add(new RowType.RowField(child.getName(), toLogicalType(child,
path + "." + child.getName())));
}
logicalType = new RowType(field.isNullable(), fields);
- } else if (arrowType instanceof ArrowType.List) {
+ } else if (arrowType instanceof ArrowType.List || arrowType instanceof
ArrowType.FixedSizeList) {
ValidationUtils.checkArgument(field.getChildren().size() == 1,
- String.format("Unsupported Arrow schema at '%s': LIST must contain
exactly one child field", path));
+ String.format("Unsupported Arrow schema at '%s': list must contain
exactly one child field", path));
Field element = field.getChildren().get(0);
logicalType = new ArrayType(field.isNullable(), toLogicalType(element,
path + "[]"));
} else {
@@ -338,11 +343,14 @@ public final class HoodieFlinkLanceArrowUtils {
return row;
}
- private static ArrayData readArray(ArrayType arrayType, ListVector vector,
int rowId) {
- int startIndex = vector.getElementStartIndex(rowId);
- int endIndex = vector.getElementEndIndex(rowId);
+ private static ArrayData readArray(ArrayType arrayType, ValueVector vector,
int rowId) {
+ BaseListVector listVector = (BaseListVector) vector;
+ FieldVector dataVector = vector instanceof FixedSizeListVector
+ ? ((FixedSizeListVector) vector).getDataVector()
+ : ((ListVector) vector).getDataVector();
+ int startIndex = listVector.getElementStartIndex(rowId);
+ int endIndex = listVector.getElementEndIndex(rowId);
Object[] values = new Object[endIndex - startIndex];
- FieldVector dataVector = vector.getDataVector();
for (int i = 0; i < values.length; i++) {
values[i] = readValue(arrayType.getElementType(), dataVector, startIndex
+ i);
}
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java
index af992d1a18f4..c0a267e32bba 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java
@@ -103,16 +103,24 @@ public class HoodieSchemaConverter {
String vectorColumns) {
Map<String, Integer> vectorColumnMap = vectorColumns == null ||
vectorColumns.trim().isEmpty()
? Collections.emptyMap() : VectorColumnParser.parse(vectorColumns);
- validateVectorColumns(logicalType, vectorColumnMap);
return convertToSchema(logicalType, rowName, vectorColumnMap);
}
- private static HoodieSchema convertToSchema(
+ /**
+ * Converts a Flink LogicalType into a HoodieSchema with the specified
top-level VECTOR columns.
+ *
+ * @param logicalType Flink logical type
+ * @param rowName the record name
+ * @param vectorColumns vector column names and dimensions
+ * @return HoodieSchema matching this logical type
+ */
+ public static HoodieSchema convertToSchema(
LogicalType logicalType,
String rowName,
Map<String, Integer> vectorColumns) {
ValidationUtils.checkArgument(vectorColumns.isEmpty() || logicalType
instanceof RowType,
"VECTOR columns can only be configured for top-level ROW schemas.");
+ validateVectorColumns(logicalType, vectorColumns);
int precision;
boolean nullable = logicalType.isNullable();
@@ -285,15 +293,15 @@ public class HoodieSchemaConverter {
* an unknown column is rejected instead of being silently ignored during
conversion.
*
* @param logicalType Flink logical type
- * @param vectorColumns parsed vector columns (normalized column name to
dimension), may be empty
+ * @param vectorColumns parsed vector columns (column name to dimension),
may be empty
*/
private static void validateVectorColumns(LogicalType logicalType,
Map<String, Integer> vectorColumns) {
if (vectorColumns.isEmpty()) {
return;
}
- List<String> normalizedFieldNames = ((RowType)
logicalType).getFieldNames();
+ List<String> fieldNames = ((RowType) logicalType).getFieldNames();
vectorColumns.keySet().stream()
- .filter(vectorColumn -> !normalizedFieldNames.contains(vectorColumn))
+ .filter(vectorColumn -> !fieldNames.contains(vectorColumn))
.findFirst()
.ifPresent(vectorColumn -> {
throw new IllegalArgumentException("VECTOR column '" + vectorColumn
+ "' does not exist in the table schema.");
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/VectorColumnParser.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/VectorColumnParser.java
index 68724be54cb6..4af139b2a05f 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/VectorColumnParser.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/VectorColumnParser.java
@@ -24,7 +24,6 @@ import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.LogicalType;
import java.util.LinkedHashMap;
-import java.util.Locale;
import java.util.Map;
/**
@@ -33,8 +32,8 @@ import java.util.Map;
*
* <p>The option uses {@code colName[:dimension]} entries separated by commas,
for example
* {@code embedding:4,features:3,codes:4}. The dimension defaults to {@value
#DEFAULT_VECTOR_DIMENSION}
- * when omitted. Column names are matched case-insensitively. The vector
element type is inferred
- * from the Flink array element type: {@code ARRAY<FLOAT>}, {@code
ARRAY<DOUBLE>}, or
+ * when omitted. The vector element type is inferred from the Flink array
element type:
+ * {@code ARRAY<FLOAT>}, {@code ARRAY<DOUBLE>}, or
* {@code ARRAY<TINYINT>} map to FLOAT, DOUBLE, and INT8 respectively.
*
* <p>The parser validates the descriptor syntax (well-formed entries, no
duplicate columns,
@@ -50,10 +49,10 @@ public class VectorColumnParser {
/**
* Parses the {@code hoodie.vector.columns} descriptor string into a map
from the
- * (lower-cased) column name to its vector dimension, preserving declaration
order.
+ * column name to its vector dimension, preserving declaration order.
*
* @param vectorColumns comma-separated {@code colName[:dimension]}
descriptors
- * @return map from normalized column name to dimension
+ * @return map from column name to dimension
* @throws IllegalArgumentException if a descriptor is malformed,
duplicated, or has a non-positive dimension
*/
public static Map<String, Integer> parse(String vectorColumns) {
@@ -69,8 +68,7 @@ public class VectorColumnParser {
"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)) {
throw new IllegalArgumentException("Duplicate VECTOR column descriptor
for column: " + columnName);
}
int dimension = DEFAULT_VECTOR_DIMENSION;
@@ -89,7 +87,7 @@ public class VectorColumnParser {
if (dimension <= 0) {
throw new IllegalArgumentException("VECTOR dimension must be positive
for column '" + columnName + "': " + dimension);
}
- parsed.put(normalizedColumnName, dimension);
+ parsed.put(columnName, dimension);
}
return parsed;
}
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/VectorConversionUtils.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/VectorConversionUtils.java
index c3721bae71ea..85488d2cd241 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/VectorConversionUtils.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/VectorConversionUtils.java
@@ -42,7 +42,6 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Locale;
import java.util.Map;
import java.util.stream.Collectors;
@@ -76,7 +75,7 @@ public final class VectorConversionUtils {
Map<String, HoodieSchema.Vector> vectorFields =
getVectorFields(tableSchema);
Map<Integer, HoodieSchema.Vector> vectorColumnInfo = new LinkedHashMap<>();
for (int i = 0; i < selectedFields.length; i++) {
- HoodieSchema.Vector vector =
vectorFields.get(fullFieldNames[selectedFields[i]].toLowerCase(Locale.ROOT));
+ HoodieSchema.Vector vector =
vectorFields.get(fullFieldNames[selectedFields[i]]);
if (vector != null) {
vectorColumnInfo.put(i, vector);
}
@@ -143,7 +142,7 @@ public final class VectorConversionUtils {
}
DataType[] readFieldTypes = Arrays.copyOf(fullFieldTypes,
fullFieldTypes.length);
for (int i = 0; i < fullFieldNames.length; i++) {
- if
(vectorFields.containsKey(fullFieldNames[i].toLowerCase(Locale.ROOT))) {
+ if (vectorFields.containsKey(fullFieldNames[i])) {
readFieldTypes[i] =
DataTypes.of(bytesType(fullFieldTypes[i].getLogicalType()));
}
}
@@ -192,13 +191,13 @@ public final class VectorConversionUtils {
}
/**
- * Returns VECTOR fields keyed by lower-cased field name.
+ * Returns VECTOR fields keyed by field name.
*/
private static Map<String, HoodieSchema.Vector> getVectorFields(HoodieSchema
tableSchema) {
return tableSchema.getFields().stream()
.filter(field -> field.schema().getNonNullType().getType() ==
HoodieSchemaType.VECTOR)
.collect(Collectors.toMap(
- field -> field.name().toLowerCase(Locale.ROOT),
+ field -> field.name(),
field -> (HoodieSchema.Vector) field.schema().getNonNullType()));
}
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java
index 22a7c711ea4e..0acfee129694 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java
@@ -560,6 +560,22 @@ public class TestHoodieSchemaConverter {
assertEquals(2, ((HoodieSchema.Vector) defaultEmbedding).getDimension());
}
+ @Test
+ public void testVectorColumnNamesAreCaseSensitive() {
+ RowType rowType = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("Embedding",
DataTypes.ARRAY(DataTypes.FLOAT().notNull()).notNull()))
+ .notNull()
+ .getLogicalType();
+
+ HoodieSchema schema = HoodieSchemaConverter.convertToSchema(rowType,
"test_record", "Embedding:4");
+ HoodieSchema vector =
schema.getField("Embedding").get().schema().getNonNullType();
+ assertEquals(HoodieSchemaType.VECTOR, vector.getType());
+ assertEquals(4, ((HoodieSchema.Vector) vector).getDimension());
+
+ assertThrows(IllegalArgumentException.class,
+ () -> HoodieSchemaConverter.convertToSchema(rowType, "test_record",
"embedding:4"));
+ }
+
@Test
public void testConvertVectorColumnsValidation() {
RowType rowType = (RowType) DataTypes.ROW(
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestVectorConversionUtils.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestVectorConversionUtils.java
index e5a2e8478fd2..fbbde80a8051 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestVectorConversionUtils.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestVectorConversionUtils.java
@@ -69,11 +69,11 @@ class TestVectorConversionUtils {
DataType requested = DataTypes.ROW(
DataTypes.FIELD("codes",
DataTypes.ARRAY(DataTypes.TINYINT()).notNull()),
DataTypes.FIELD("id", DataTypes.INT().notNull()),
- DataTypes.FIELD("float_vec",
DataTypes.ARRAY(DataTypes.FLOAT()))).notNull();
+ DataTypes.FIELD("Float_Vec",
DataTypes.ARRAY(DataTypes.FLOAT()))).notNull();
HoodieSchema projectedSchema = HoodieSchema.createRecord("projected",
null, null, Arrays.asList(
HoodieSchemaField.of("codes", INT8_VECTOR),
HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.INT)),
- HoodieSchemaField.of("float_vec", FLOAT_VECTOR)));
+ HoodieSchemaField.of("Float_Vec", FLOAT_VECTOR)));
DataType physical = VectorConversionUtils.getParquetReadDataType(
requested, projectedSchema, detected);
RowType physicalRow = (RowType) physical.getLogicalType();
@@ -92,6 +92,14 @@ class TestVectorConversionUtils {
assertEquals(LogicalTypeRoot.VARBINARY,
physicalTypes[1].getLogicalType().getTypeRoot());
assertEquals(LogicalTypeRoot.VARBINARY,
physicalTypes[2].getLogicalType().getTypeRoot());
assertEquals(LogicalTypeRoot.VARBINARY,
physicalTypes[3].getLogicalType().getTypeRoot());
+
+ String[] mismatchedNames = {"id", "float_vec", "double_vec", "codes"};
+ Map<Integer, HoodieSchema.Vector> mismatched =
+ VectorConversionUtils.detectVectorColumns(mismatchedNames, selected,
schema);
+ assertFalse(mismatched.containsKey(2));
+ DataType[] mismatchedPhysicalTypes =
+ VectorConversionUtils.getParquetReadFieldTypes(mismatchedNames,
fieldTypes, schema);
+ assertEquals(LogicalTypeRoot.ARRAY,
mismatchedPhysicalTypes[1].getLogicalType().getTypeRoot());
}
@Test
@@ -149,7 +157,7 @@ class TestVectorConversionUtils {
private static HoodieSchema vectorRecordSchema() {
return HoodieSchema.createRecord("vectors", null, null, Arrays.asList(
HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.INT)),
- HoodieSchemaField.of("float_vec", FLOAT_VECTOR),
+ HoodieSchemaField.of("Float_Vec", FLOAT_VECTOR),
HoodieSchemaField.of("double_vec", DOUBLE_VECTOR),
HoodieSchemaField.of("codes", INT8_VECTOR)));
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java
index ac067d36f412..636a79d00440 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java
@@ -87,15 +87,22 @@ public class FormatUtils {
List<DataType> fieldTypes,
int[] selectedFields,
Configuration hadoopConf) {
- DataType selectedDataType = DataTypes.ROW(Arrays.stream(selectedFields)
+ HoodieSchema requestedSchema = HoodieSchemaConverter.convertToSchema(
+ DataTypes.ROW(Arrays.stream(selectedFields)
.mapToObj(i -> DataTypes.FIELD(fieldNames.get(i),
fieldTypes.get(i)))
.toArray(DataTypes.Field[]::new))
- .bridgedTo(RowData.class);
- HoodieSchema requestedSchema =
HoodieSchemaConverter.convertToSchema(selectedDataType.getLogicalType());
+ .getLogicalType());
+ return getLanceRecordIterator(path, requestedSchema, hadoopConf);
+ }
+
+ public static ClosableIterator<RowData> getLanceRecordIterator(
+ String path,
+ HoodieSchema requestedSchema,
+ Configuration hadoopConf) {
HoodieRowDataLanceReader reader = new HoodieRowDataLanceReader(
new StoragePath(path), StreamerUtil.getLanceReadConfig(hadoopConf));
try {
- return reader.getRowDataIterator(selectedDataType, requestedSchema);
+ return reader.getRowDataIterator(requestedSchema);
} catch (RuntimeException e) {
reader.close();
throw new HoodieException("Failed to get iterator from Lance reader: " +
path, e);
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieRowDataLanceReader.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieRowDataLanceReader.java
index 02d1f3c073f3..27b13af8fadc 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieRowDataLanceReader.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieRowDataLanceReader.java
@@ -26,6 +26,8 @@ import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaField;
+import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.common.util.collection.CloseableMappingIterator;
@@ -44,9 +46,11 @@ import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.vector.FieldVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.arrow.vector.types.FloatingPointPrecision;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
import org.apache.arrow.vector.types.pojo.Schema;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.lance.file.LanceFileReader;
@@ -54,6 +58,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -142,14 +147,14 @@ public class HoodieRowDataLanceReader implements
HoodieRowDataFileReader {
@Override
public ClosableIterator<HoodieRecord<RowData>>
getRecordIterator(HoodieSchema readerSchema, HoodieSchema requestedSchema)
throws IOException {
- ClosableIterator<RowData> rowDataItr =
getRowDataIterator(RowDataQueryContexts.fromSchema(requestedSchema).getRowType(),
requestedSchema);
+ ClosableIterator<RowData> rowDataItr = getRowDataIterator(requestedSchema);
return new CloseableMappingIterator<>(rowDataItr, HoodieFlinkRecord::new);
}
@Override
public ClosableIterator<String> getRecordKeyIterator() throws IOException {
HoodieSchema schema = HoodieSchemaUtils.getRecordKeySchema();
- ClosableIterator<RowData> rowDataItr =
getRowDataIterator(RowDataQueryContexts.fromSchema(schema).getRowType(),
schema);
+ ClosableIterator<RowData> rowDataItr = getRowDataIterator(schema);
return new CloseableMappingIterator<>(rowDataItr, rowData ->
rowData.getString(0).toString());
}
@@ -164,11 +169,13 @@ public class HoodieRowDataLanceReader implements
HoodieRowDataFileReader {
&&
!internalSchemaManager.getMergeSchema(path.getName()).isEmptySchema()) {
throw new HoodieValidationException("Flink Lance base-file support does
not support schema evolution.");
}
- return
getRowDataIterator(RowDataQueryContexts.fromSchema(requiredSchema).getRowType(),
requiredSchema);
+ return getRowDataIterator(requiredSchema);
}
- public ClosableIterator<RowData> getRowDataIterator(DataType dataType,
HoodieSchema requestedSchema) {
- RowType rowType = (RowType) dataType.getLogicalType();
+ public ClosableIterator<RowData> getRowDataIterator(HoodieSchema
requestedSchema) {
+ validateRequestedVectors(requestedSchema);
+ RowType rowType = (RowType)
RowDataQueryContexts.fromSchema(requestedSchema)
+ .getRowType().getLogicalType();
List<String> columnNames = new ArrayList<>(rowType.getFieldCount());
for (RowType.RowField field : rowType.getFields()) {
columnNames.add(field.getName());
@@ -204,7 +211,64 @@ public class HoodieRowDataLanceReader implements
HoodieRowDataFileReader {
@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);
+ }
+ Map<String, Integer> vectorColumns = new LinkedHashMap<>();
+ vectorColumnNames.forEach(name -> vectorColumns.put(
+ name,
vectorSchemaFromField(arrowSchema.findField(name)).getDimension()));
+ 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 =
vectorSchemaFromField(arrowSchema.findField(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 static HoodieSchema.Vector vectorSchemaFromField(Field field) {
+ // Spark Lance currently writes VECTOR columns as FixedSizeList only for
FLOAT/DOUBLE.
+ // Restrict restoration to the same types so files have one cross-engine
VECTOR contract.
+ if (!(field.getType() instanceof ArrowType.FixedSizeList)
+ || field.getChildren().size() != 1
+ || !(field.getChildren().get(0).getType() instanceof
ArrowType.FloatingPoint)) {
+ throw new HoodieValidationException(
+ "Invalid Lance VECTOR encoding for column '" + field.getName()
+ + "': expected FixedSizeList<Float32|Float64> but found " +
field);
+ }
+
+ ArrowType.FixedSizeList listType = (ArrowType.FixedSizeList)
field.getType();
+ FloatingPointPrecision precision =
+ ((ArrowType.FloatingPoint)
field.getChildren().get(0).getType()).getPrecision();
+ HoodieSchema.Vector.VectorElementType vectorElementType;
+ switch (precision) {
+ case SINGLE:
+ vectorElementType = HoodieSchema.Vector.VectorElementType.FLOAT;
+ break;
+ case DOUBLE:
+ vectorElementType = HoodieSchema.Vector.VectorElementType.DOUBLE;
+ break;
+ default:
+ throw new HoodieValidationException(
+ "Invalid Lance VECTOR encoding for column '" + field.getName()
+ + "': expected Float32 or Float64 elements but found " +
precision);
+ }
+ return HoodieSchema.createVector(listType.getListSize(),
vectorElementType);
}
@Override
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java
index a4d1be2258c9..33de1db7125a 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java
@@ -27,6 +27,7 @@ import org.apache.hudi.table.format.FilePathUtils;
import org.apache.hudi.table.format.FormatUtils;
import org.apache.hudi.table.format.InternalSchemaManager;
import org.apache.hudi.table.format.RecordIterators;
+import org.apache.hudi.util.DataTypeUtils;
import org.apache.hudi.util.VectorConversionUtils;
import lombok.extern.slf4j.Slf4j;
@@ -37,8 +38,10 @@ import
org.apache.flink.api.common.io.compression.InflaterInputStreamFactory;
import org.apache.flink.core.fs.FileInputSplit;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
@@ -76,6 +79,7 @@ public class CopyOnWriteInputFormat extends
FileInputFormat<RowData> {
private final DataType[] readFieldTypes;
private final int[] selectedFields;
private final Map<Integer, HoodieSchema.Vector> vectorColumnInfo;
+ private final HoodieSchema tableSchema;
private final String partDefaultName;
private final String partPathField;
private final boolean hiveStylePartitioning;
@@ -86,6 +90,7 @@ public class CopyOnWriteInputFormat extends
FileInputFormat<RowData> {
private transient ClosableIterator<RowData> itr;
private transient long currentReadCount;
+ private transient HoodieSchema requestedSchema;
/**
* Files filter for determining what files/directories should be included.
@@ -119,6 +124,7 @@ public class CopyOnWriteInputFormat extends
FileInputFormat<RowData> {
this.readFieldTypes =
VectorConversionUtils.getParquetReadFieldTypes(fullFieldNames, fullFieldTypes,
tableSchema);
this.selectedFields = selectedFields;
this.vectorColumnInfo =
VectorConversionUtils.detectVectorColumns(fullFieldNames, selectedFields,
tableSchema);
+ this.tableSchema = tableSchema;
this.conf = new SerializableConfiguration(conf);
this.utcTimestamp = utcTimestamp;
this.internalSchemaManager = internalSchemaManager;
@@ -158,8 +164,7 @@ public class CopyOnWriteInputFormat extends
FileInputFormat<RowData> {
}
private ClosableIterator<RowData> getLanceRecordIterator(Path path) {
- return FormatUtils.getLanceRecordIterator(
- path.toString(), Arrays.asList(fullFieldNames),
Arrays.asList(fullFieldTypes), selectedFields, conf.conf());
+ return FormatUtils.getLanceRecordIterator(path.toString(),
getRequestedSchema(), conf.conf());
}
@Override
@@ -419,4 +424,15 @@ public class CopyOnWriteInputFormat extends
FileInputFormat<RowData> {
}
}
+ private HoodieSchema getRequestedSchema() {
+ if (requestedSchema == null) {
+ RowType requestedRowType = (RowType)
DataTypes.ROW(Arrays.stream(selectedFields)
+ .mapToObj(i -> DataTypes.FIELD(fullFieldNames[i],
fullFieldTypes[i]))
+ .toArray(DataTypes.Field[]::new))
+ .notNull()
+ .getLogicalType();
+ requestedSchema = DataTypeUtils.toHoodieSchema(requestedRowType,
tableSchema);
+ }
+ return requestedSchema;
+ }
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestVectorDataSource.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestVectorDataSource.java
index 77416c59184f..d97e5a6994b8 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestVectorDataSource.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestVectorDataSource.java
@@ -51,6 +51,7 @@ import org.apache.hudi.util.HoodieSchemaConverter;
import org.apache.hudi.util.StreamerUtil;
import org.apache.hudi.utils.FlinkMiniCluster;
import org.apache.hudi.utils.TestTableEnvs;
+import org.apache.hudi.utils.TestUtils;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
@@ -166,6 +167,65 @@ public class ITTestVectorDataSource {
assertFloatArray(vectorProjection.get(0).getField(1), new float[] {-1.0f,
-2.0f, -3.0f, -4.0f});
}
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ public void testLanceVectorUpsertRead(HoodieTableType tableType) throws
Exception {
+ TableEnvironment tableEnv = TestTableEnvs.getBatchTableEnv();
+ String tablePath = tempDir.resolve("lance_upsert_" +
tableType.name()).toUri().toString();
+ Map<String, String> extraOptions = new LinkedHashMap<>();
+ extraOptions.put("hoodie.table.base.file.format", "LANCE");
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ extraOptions.put(FlinkOptions.COMPACTION_SCHEDULE_ENABLED.key(), "true");
+ extraOptions.put(FlinkOptions.COMPACTION_ASYNC_ENABLED.key(), "true");
+ extraOptions.put(FlinkOptions.COMPACTION_DELTA_COMMITS.key(), "1");
+ }
+ createVectorTable(
+ tableEnv,
+ "vector_table",
+ tablePath,
+ tableType,
+ "embedding:2,features:3,nullable_embedding:2",
+ null,
+ extraOptions);
+
+ execInsertSql(tableEnv,
+ "INSERT INTO vector_table(id, embedding, features, nullable_embedding,
label, tags, ts) VALUES "
+ + "('id1', ARRAY[CAST(1.0 AS FLOAT), CAST(1.5 AS FLOAT)], "
+ + " ARRAY[CAST(10.0 AS DOUBLE), CAST(10.5 AS DOUBLE), CAST(11.0 AS
DOUBLE)], "
+ + " ARRAY[CAST(7.0 AS FLOAT), CAST(7.5 AS FLOAT)], 'old1',
ARRAY['red', 'blue'], 1), "
+ + "('id2', ARRAY[CAST(2.0 AS FLOAT), CAST(2.5 AS FLOAT)], "
+ + " ARRAY[CAST(20.0 AS DOUBLE), CAST(20.5 AS DOUBLE), CAST(21.0 AS
DOUBLE)], "
+ + " CAST(NULL AS ARRAY<FLOAT>), 'old2', ARRAY['green'], 2)");
+ execInsertSql(tableEnv,
+ "INSERT INTO vector_table(id, embedding, features, nullable_embedding,
label, tags, ts) VALUES "
+ + "('id1', ARRAY[CAST(9.0 AS FLOAT), CAST(9.5 AS FLOAT)], "
+ + " ARRAY[CAST(90.0 AS DOUBLE), CAST(90.5 AS DOUBLE), CAST(91.0 AS
DOUBLE)], "
+ + " CAST(NULL AS ARRAY<FLOAT>), 'new1', ARRAY['black'], 10)");
+
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ assertTrue(TestUtils.hasCompleteCompactionInstant(tablePath));
+ }
+
+ List<Row> rows = collect(tableEnv,
+ "SELECT tags, features, id, nullable_embedding, embedding, label "
+ + "FROM vector_table ORDER BY id");
+ assertEquals(2, rows.size());
+
+ assertObjectArray(rows.get(0).getField(0), new Object[] {"black"});
+ assertDoubleArray(rows.get(0).getField(1), new double[] {90.0D, 90.5D,
91.0D});
+ assertEquals("id1", rows.get(0).getField(2));
+ assertNull(rows.get(0).getField(3));
+ assertFloatArray(rows.get(0).getField(4), new float[] {9.0F, 9.5F});
+ assertEquals("new1", rows.get(0).getField(5));
+
+ assertObjectArray(rows.get(1).getField(0), new Object[] {"green"});
+ assertDoubleArray(rows.get(1).getField(1), new double[] {20.0D, 20.5D,
21.0D});
+ assertEquals("id2", rows.get(1).getField(2));
+ assertNull(rows.get(1).getField(3));
+ assertFloatArray(rows.get(1).getField(4), new float[] {2.0F, 2.5F});
+ assertEquals("old2", rows.get(1).getField(5));
+ }
+
@Test
public void testColumnProjectionContainsVectorColumn() throws Exception {
TableEnvironment tableEnv = TestTableEnvs.getBatchTableEnv();
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestHoodieRowDataLanceReader.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestHoodieRowDataLanceReader.java
index 453cdbe4d508..fc39ea84fb3c 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestHoodieRowDataLanceReader.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestHoodieRowDataLanceReader.java
@@ -21,14 +21,20 @@ package org.apache.hudi.table.format;
import org.apache.hudi.common.bloom.SimpleBloomFilter;
import org.apache.hudi.common.config.HoodieConfig;
+import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaField;
+import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.schema.internal.InternalSchema;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieValidationException;
+import org.apache.hudi.io.storage.row.HoodieRowDataLanceWriter;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.util.HoodieSchemaConverter;
import org.apache.hudi.util.RowDataQueryContexts;
import org.apache.arrow.memory.BufferAllocator;
@@ -36,15 +42,30 @@ import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VarCharVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.arrow.vector.types.FloatingPointPrecision;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DoubleType;
+import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import org.lance.file.LanceFileReader;
import org.mockito.MockedStatic;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -77,6 +98,177 @@ import static org.mockito.Mockito.when;
class TestHoodieRowDataLanceReader {
private static final StoragePath PATH = new StoragePath("/tmp/test.lance");
+ @TempDir
+ Path tempDir;
+
+ @Test
+ void testRestoresMixedCaseVectorFieldName() throws Exception {
+ HoodieSchema hoodieSchema = HoodieSchema.createRecord(
+ "mixed_case_record",
+ null,
+ null,
+ Collections.singletonList(HoodieSchemaField.of(
+ "Embedding",
+ HoodieSchema.createNullable(HoodieSchema.createVector(
+ 2, HoodieSchema.Vector.VectorElementType.FLOAT)),
+ null,
+ HoodieSchema.NULL_VALUE)));
+ StoragePath path = new
StoragePath(tempDir.resolve("mixed-case-vector.lance").toUri());
+
+ try (HoodieRowDataLanceWriter writer = new HoodieRowDataLanceWriter(
+ path,
+ hoodieSchema,
+ "001",
+ mock(TaskContextSupplier.class),
+ Option.empty(),
+ 128 * 1024 * 1024L,
+ 64 * 1024 * 1024L,
+ 16 * 1024 * 1024L,
+ true,
+ false,
+ false)) {
+ writer.writeRow("key1", GenericRowData.of(
+ new GenericArrayData(new Object[] {1.25F, 2.5F})));
+ }
+
+ try (HoodieRowDataLanceReader reader = new HoodieRowDataLanceReader(path,
new HoodieConfig())) {
+ HoodieSchema restoredSchema = reader.getSchema().getNonNullType();
+ assertTrue(restoredSchema.getField("Embedding").isPresent());
+ assertFalse(restoredSchema.getField("embedding").isPresent());
+ HoodieSchema.Vector vector = (HoodieSchema.Vector)
restoredSchema.getField("Embedding")
+ .get().schema().getNonNullType();
+ assertEquals(2, vector.getDimension());
+ assertEquals(HoodieSchema.Vector.VectorElementType.FLOAT,
vector.getVectorElementType());
+ }
+ }
+
+ @Test
+ void testReadsVectorsAndRestoresSchemaIdentity() throws Exception {
+ RowType rowType = RowType.of(
+ new LogicalType[] {
+ new IntType(false),
+ new ArrayType(true, new FloatType(false)),
+ new ArrayType(false, new DoubleType(false)),
+ new ArrayType(false, new IntType(false))
+ },
+ new String[] {"id", "embedding", "features", "values"});
+ HoodieSchema hoodieSchema = HoodieSchemaConverter.convertToSchema(
+ rowType, "vector_record", "embedding:2,features:3");
+ StoragePath path = new
StoragePath(tempDir.resolve("vectors.lance").toUri());
+
+ try (HoodieRowDataLanceWriter writer = new HoodieRowDataLanceWriter(
+ path,
+ hoodieSchema,
+ "001",
+ mock(TaskContextSupplier.class),
+ Option.empty(),
+ 128 * 1024 * 1024L,
+ 64 * 1024 * 1024L,
+ 16 * 1024 * 1024L,
+ true,
+ false,
+ false)) {
+ writer.writeRow("key1", GenericRowData.of(
+ 1,
+ new GenericArrayData(new Object[] {1.25F, 2.5F}),
+ new GenericArrayData(new Object[] {3.5D, 4.5D, 5.5D}),
+ new GenericArrayData(new Object[] {10, 20})));
+ writer.writeRow("key2", GenericRowData.of(
+ 2,
+ null,
+ new GenericArrayData(new Object[] {6.5D, 7.5D, 8.5D}),
+ new GenericArrayData(new Object[] {30})));
+ }
+
+ try (HoodieRowDataLanceReader reader = new HoodieRowDataLanceReader(path,
new HoodieConfig())) {
+ HoodieSchema readSchema = reader.getSchema().getNonNullType();
+ HoodieSchema.Vector floatVector = (HoodieSchema.Vector)
readSchema.getField("embedding")
+ .get().schema().getNonNullType();
+ HoodieSchema.Vector doubleVector = (HoodieSchema.Vector)
readSchema.getField("features")
+ .get().schema().getNonNullType();
+ assertEquals(HoodieSchemaType.VECTOR, floatVector.getType());
+ assertEquals(2, floatVector.getDimension());
+ assertEquals(HoodieSchema.Vector.VectorElementType.FLOAT,
floatVector.getVectorElementType());
+ assertEquals(3, doubleVector.getDimension());
+ assertEquals(HoodieSchema.Vector.VectorElementType.DOUBLE,
doubleVector.getVectorElementType());
+ assertEquals(HoodieSchemaType.ARRAY,
+
readSchema.getField("values").get().schema().getNonNullType().getType());
+
+ try (ClosableIterator<RowData> rows =
reader.getRowDataIterator(hoodieSchema)) {
+ RowData first = rows.next();
+ assertEquals(1, first.getInt(0));
+ assertFloatArray(first.getArray(1), 1.25F, 2.5F);
+ assertDoubleArray(first.getArray(2), 3.5D, 4.5D, 5.5D);
+ assertIntArray(first.getArray(3), 10, 20);
+
+ RowData second = rows.next();
+ assertEquals(2, second.getInt(0));
+ assertTrue(second.isNullAt(1));
+ assertDoubleArray(second.getArray(2), 6.5D, 7.5D, 8.5D);
+ assertIntArray(second.getArray(3), 30);
+ assertFalse(rows.hasNext());
+ }
+ }
+
+ RowType projectedRowType = RowType.of(
+ new LogicalType[] {
+ new ArrayType(false, new IntType(false)),
+ new ArrayType(false, new DoubleType(false)),
+ new IntType(false),
+ new ArrayType(true, new FloatType(false))
+ },
+ new String[] {"values", "features", "id", "embedding"});
+ HoodieSchema projectedSchema = HoodieSchemaConverter.convertToSchema(
+ projectedRowType, "projected_record", "features:3,embedding:2");
+ try (HoodieRowDataLanceReader reader = new HoodieRowDataLanceReader(path,
new HoodieConfig());
+ ClosableIterator<RowData> rows =
reader.getRowDataIterator(projectedSchema)) {
+ RowData first = rows.next();
+ assertIntArray(first.getArray(0), 10, 20);
+ assertDoubleArray(first.getArray(1), 3.5D, 4.5D, 5.5D);
+ assertEquals(1, first.getInt(2));
+ assertFloatArray(first.getArray(3), 1.25F, 2.5F);
+ }
+
+ RowType incompatibleRowType = RowType.of(
+ new LogicalType[] {new ArrayType(true, new FloatType(false))},
+ new String[] {"embedding"});
+ HoodieSchema incompatibleSchema = HoodieSchemaConverter.convertToSchema(
+ incompatibleRowType, "incompatible_record", "embedding:3");
+ try (HoodieRowDataLanceReader reader = new HoodieRowDataLanceReader(path,
new HoodieConfig())) {
+ HoodieValidationException exception = assertThrows(
+ HoodieValidationException.class,
+ () -> reader.getRowDataIterator(incompatibleSchema));
+ assertTrue(exception.getMessage().contains("requested VECTOR(3)"));
+ assertTrue(exception.getMessage().contains("file contains VECTOR(2)"));
+ }
+ }
+
+ @Test
+ void testRejectsInvalidVectorArrowTypes() throws Exception {
+ RowType rowType = RowType.of(
+ new LogicalType[] {new ArrayType(false, new FloatType(false))},
+ new String[] {"embedding"});
+ HoodieSchema requestedSchema = HoodieSchemaConverter.convertToSchema(
+ rowType, "vector_record", "embedding:2");
+
+ Field element = new Field(
+ "element", FieldType.notNullable(new
ArrowType.FloatingPoint(FloatingPointPrecision.SINGLE)), null);
+ assertInvalidVectorEncoding(
+ requestedSchema,
+ new Field("embedding", FieldType.nullable(new ArrowType.List()),
Collections.singletonList(element)),
+ "expected FixedSizeList<Float32|Float64>");
+
+ Field halfElement = new Field(
+ "element", FieldType.notNullable(new
ArrowType.FloatingPoint(FloatingPointPrecision.HALF)), null);
+ assertInvalidVectorEncoding(
+ requestedSchema,
+ new Field(
+ "embedding",
+ FieldType.nullable(new ArrowType.FixedSizeList(2)),
+ Collections.singletonList(halfElement)),
+ "expected Float32 or Float64 elements");
+ }
+
@Test
void testReadsMetadataAndClosesIdempotently() throws Exception {
SimpleBloomFilter bloomFilter = new SimpleBloomFilter(100, 0.01,
MURMUR_HASH);
@@ -196,19 +388,52 @@ class TestHoodieRowDataLanceReader {
try (MockedStatic<LanceFileReader> mocked = mockLanceOpen(metadataReader,
dataReader)) {
HoodieRowDataLanceReader reader = new HoodieRowDataLanceReader(PATH, new
HoodieConfig());
- assertThrows(HoodieException.class, () -> reader.getRowDataIterator(
- RowDataQueryContexts.fromSchema(schema).getRowType(), schema));
+ assertThrows(HoodieException.class, () ->
reader.getRowDataIterator(schema));
verify(dataReader).close();
reader.close();
}
}
+ private static void assertFloatArray(ArrayData array, float... expected) {
+ assertEquals(expected.length, array.size());
+ for (int i = 0; i < expected.length; i++) {
+ assertEquals(expected[i], array.getFloat(i));
+ }
+ }
+
+ private static void assertDoubleArray(ArrayData array, double... expected) {
+ assertEquals(expected.length, array.size());
+ for (int i = 0; i < expected.length; i++) {
+ assertEquals(expected[i], array.getDouble(i));
+ }
+ }
+
+ private static void assertIntArray(ArrayData array, int... expected) {
+ assertEquals(expected.length, array.size());
+ for (int i = 0; i < expected.length; i++) {
+ assertEquals(expected[i], array.getInt(i));
+ }
+ }
+
private static LanceFileReader metadataReader() throws Exception {
LanceFileReader reader = mock(LanceFileReader.class);
when(reader.schema()).thenReturn(new Schema(Collections.emptyList()));
return reader;
}
+ private static void assertInvalidVectorEncoding(
+ HoodieSchema requestedSchema, Field field, String expectedMessage)
throws Exception {
+ LanceFileReader metadataReader = mock(LanceFileReader.class);
+ when(metadataReader.schema()).thenReturn(new
Schema(Collections.singletonList(field)));
+ try (MockedStatic<LanceFileReader> mocked = mockLanceOpen(metadataReader);
+ HoodieRowDataLanceReader reader = new HoodieRowDataLanceReader(PATH,
new HoodieConfig())) {
+ HoodieValidationException exception = assertThrows(
+ HoodieValidationException.class,
+ () -> reader.getRowDataIterator(requestedSchema));
+ assertTrue(exception.getMessage().contains(expectedMessage));
+ }
+ }
+
private static MockedStatic<LanceFileReader>
mockLanceOpen(LanceFileReader... readers) {
MockedStatic<LanceFileReader> mocked = mockStatic(LanceFileReader.class);
AtomicInteger readerIndex = new AtomicInteger();
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cow/TestCopyOnWriteInputFormat.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cow/TestCopyOnWriteInputFormat.java
index c8210d170652..3355888c5c5d 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cow/TestCopyOnWriteInputFormat.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cow/TestCopyOnWriteInputFormat.java
@@ -19,8 +19,15 @@
package org.apache.hudi.table.format.cow;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieValidationException;
+import org.apache.hudi.io.storage.row.HoodieRowDataLanceWriter;
+import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.format.InternalSchemaManager;
import org.apache.hudi.util.HoodieSchemaConverter;
import org.apache.hudi.utils.TestConfigurations;
@@ -28,10 +35,15 @@ import org.apache.hudi.utils.TestConfigurations;
import org.apache.flink.api.common.io.FilePathFilter;
import org.apache.flink.core.fs.FileInputSplit;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.data.GenericArrayData;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
import org.apache.hadoop.fs.FileStatus;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -103,6 +115,57 @@ class TestCopyOnWriteInputFormat {
assertEquals(-1L, splits[0].getLength());
}
+ @Test
+ void testLanceVectorSchemaMismatch() throws Exception {
+ RowType rowType = RowType.of(false,
+ new LogicalType[] {new ArrayType(false, new FloatType(false))}, new
String[] {"embedding"});
+ HoodieSchema fileSchema = HoodieSchemaConverter.convertToSchema(
+ rowType, "file_record", "embedding:2");
+ HoodieSchema tableSchema = HoodieSchemaConverter.convertToSchema(
+ rowType, "table_record", "embedding:3");
+ StoragePath storagePath = new
StoragePath(tempDir.resolve("vector.lance").toUri());
+
+ try (HoodieRowDataLanceWriter writer = new HoodieRowDataLanceWriter(
+ storagePath,
+ fileSchema,
+ "001",
+ mock(TaskContextSupplier.class),
+ Option.empty(),
+ 128 * 1024 * 1024L,
+ 64 * 1024 * 1024L,
+ 16 * 1024 * 1024L,
+ true,
+ false,
+ false)) {
+ writer.writeRow("key1", GenericRowData.of(
+ new GenericArrayData(new Object[] {1.0F, 2.0F})));
+ }
+
+ DataType rowDataType =
HoodieSchemaConverter.convertToDataType(tableSchema);
+ CopyOnWriteInputFormat inputFormat = new CopyOnWriteInputFormat(
+ new Path[] {new Path(storagePath.toUri())},
+ new String[] {"embedding"},
+ rowDataType.getChildren().toArray(new DataType[0]),
+ new int[] {0},
+ FlinkOptions.PARTITION_DEFAULT_NAME.defaultValue(),
+ FlinkOptions.PARTITION_PATH_FIELD.defaultValue(),
+ false,
+ Collections.emptyList(),
+ Long.MAX_VALUE,
+ new org.apache.hadoop.conf.Configuration(),
+ true,
+ InternalSchemaManager.DISABLED,
+ tableSchema);
+
+ HoodieException exception = assertThrows(
+ HoodieException.class,
+ () -> inputFormat.open(new FileInputSplit(
+ 0, new Path(storagePath.toUri()), 0, -1, new String[0])));
+ assertTrue(exception.getCause() instanceof HoodieValidationException);
+ assertTrue(exception.getCause().getMessage().contains("requested
VECTOR(3)"));
+ assertTrue(exception.getCause().getMessage().contains("file contains
VECTOR(2)"));
+ }
+
@Test
void testAcceptFileUsesBuiltInAndCustomFilters() {
CopyOnWriteInputFormat inputFormat = inputFormat(
