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 29a4925a0f01 feat(flink): support nested data type for flink lance
reader/writer (#19242)
29a4925a0f01 is described below
commit 29a4925a0f01f121f17623e19a9afa562bcef567
Author: Shuo Cheng <[email protected]>
AuthorDate: Tue Jul 14 12:44:17 2026 +0800
feat(flink): support nested data type for flink lance reader/writer (#19242)
* feat(flink): support nested data type for flink lance reader/writer
---
.../io/storage/row/HoodieRowDataLanceWriter.java | 9 +-
.../{ => lance}/HoodieFlinkLanceArrowUtils.java | 183 ++++----
.../io/storage/row/lance/LanceRowDataWriter.java | 482 +++++++++++++++++++++
.../row/TestHoodieFlinkLanceArrowUtils.java | 58 ++-
.../io/storage/row/TestLanceRowDataWriter.java | 274 ++++++++++++
.../table/format/HoodieRowDataLanceReader.java | 2 +-
.../apache/hudi/table/ITTestHoodieDataSource.java | 43 ++
7 files changed, 937 insertions(+), 114 deletions(-)
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataLanceWriter.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataLanceWriter.java
index 6381dba3005f..0990e3d6527a 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataLanceWriter.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataLanceWriter.java
@@ -26,6 +26,8 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.io.lance.HoodieBaseLanceWriter;
+import org.apache.hudi.io.storage.row.lance.HoodieFlinkLanceArrowUtils;
+import org.apache.hudi.io.storage.row.lance.LanceRowDataWriter;
import org.apache.hudi.storage.StoragePath;
import org.apache.arrow.vector.VectorSchemaRoot;
@@ -134,18 +136,17 @@ public class HoodieRowDataLanceWriter extends
HoodieBaseLanceWriter<RowData, Str
private class RowDataArrowWriter implements ArrowWriter<RowData> {
private final VectorSchemaRoot root;
+ private final LanceRowDataWriter writer;
private int rowId;
private RowDataArrowWriter(VectorSchemaRoot root) {
this.root = root;
+ this.writer = new LanceRowDataWriter(rowType, root.getFieldVectors(),
utcTimestamp);
}
@Override
public void write(RowData row) {
- for (int i = 0; i < rowType.getFieldCount(); i++) {
- HoodieFlinkLanceArrowUtils.writeValue(rowType.getTypeAt(i),
root.getVector(i), rowId, row, i, utcTimestamp);
- }
- rowId++;
+ writer.write(row, rowId++);
}
@Override
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieFlinkLanceArrowUtils.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/lance/HoodieFlinkLanceArrowUtils.java
similarity index 66%
rename from
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieFlinkLanceArrowUtils.java
rename to
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/lance/HoodieFlinkLanceArrowUtils.java
index 07171f615b46..6c4e4ac29cb0 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieFlinkLanceArrowUtils.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/lance/HoodieFlinkLanceArrowUtils.java
@@ -16,8 +16,9 @@
* limitations under the License.
*/
-package org.apache.hudi.io.storage.row;
+package org.apache.hudi.io.storage.row.lance;
+import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.arrow.vector.BigIntVector;
@@ -30,11 +31,13 @@ import org.apache.arrow.vector.Float8Vector;
import org.apache.arrow.vector.IntVector;
import org.apache.arrow.vector.SmallIntVector;
import org.apache.arrow.vector.TimeMilliVector;
-import org.apache.arrow.vector.TimeStampMicroVector;
+import org.apache.arrow.vector.TimeStampVector;
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.ListVector;
+import org.apache.arrow.vector.complex.StructVector;
import org.apache.arrow.vector.types.DateUnit;
import org.apache.arrow.vector.types.FloatingPointPrecision;
import org.apache.arrow.vector.types.TimeUnit;
@@ -42,11 +45,14 @@ 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.DecimalData;
+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.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.BooleanType;
import org.apache.flink.table.types.logical.DateType;
@@ -66,13 +72,16 @@ import org.apache.flink.table.types.logical.VarCharType;
import java.math.BigDecimal;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
-import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getPrecision;
-
/**
- * Primitive RowData/Arrow conversion helpers for Flink Lance base files.
+ * RowData/Arrow conversion helpers for Flink Lance base files.
+ *
+ * <p>Supports primitive types, {@code ROW}, and {@code ARRAY}. Currently,
{@code MAP} is not
+ * supported in the current Lance version. Enable it once the lance version is
upgraded to support
+ * {@code MAP}.
+ *
+ * @see <a href="https://github.com/lance-format/lance/issues/3620";>Lance
issue #3620</a>
*/
public final class HoodieFlinkLanceArrowUtils {
@@ -90,7 +99,7 @@ public final class HoodieFlinkLanceArrowUtils {
public static RowType toRowType(Schema schema) {
List<RowType.RowField> fields = new ArrayList<>(schema.getFields().size());
for (Field field : schema.getFields()) {
- fields.add(new RowType.RowField(field.getName(),
toLogicalType(field.getType())));
+ fields.add(new RowType.RowField(field.getName(), toLogicalType(field,
field.getName())));
}
return new RowType(fields);
}
@@ -108,68 +117,10 @@ public final class HoodieFlinkLanceArrowUtils {
return rowData;
}
- public static void writeValue(LogicalType type, FieldVector vector, int
rowId, RowData rowData, int ordinal) {
- writeValue(type, vector, rowId, rowData, ordinal, true);
- }
-
- public static void writeValue(LogicalType type, FieldVector vector, int
rowId, RowData rowData, int ordinal, boolean utcTimestamp) {
- if (rowData.isNullAt(ordinal)) {
- vector.setNull(rowId);
- return;
- }
- switch (type.getTypeRoot()) {
- case BOOLEAN:
- ((BitVector) vector).setSafe(rowId, rowData.getBoolean(ordinal) ? 1 :
0);
- return;
- case TINYINT:
- ((TinyIntVector) vector).setSafe(rowId, rowData.getByte(ordinal));
- return;
- case SMALLINT:
- ((SmallIntVector) vector).setSafe(rowId, rowData.getShort(ordinal));
- return;
- case INTEGER:
- ((IntVector) vector).setSafe(rowId, rowData.getInt(ordinal));
- return;
- case DATE:
- ((DateDayVector) vector).setSafe(rowId, rowData.getInt(ordinal));
- return;
- case TIME_WITHOUT_TIME_ZONE:
- ((TimeMilliVector) vector).setSafe(rowId, rowData.getInt(ordinal));
- return;
- case BIGINT:
- ((BigIntVector) vector).setSafe(rowId, rowData.getLong(ordinal));
- return;
- case FLOAT:
- ((Float4Vector) vector).setSafe(rowId, rowData.getFloat(ordinal));
- return;
- case DOUBLE:
- ((Float8Vector) vector).setSafe(rowId, rowData.getDouble(ordinal));
- return;
- case CHAR:
- case VARCHAR:
- ((VarCharVector) vector).setSafe(rowId,
rowData.getString(ordinal).toBytes());
- return;
- case BINARY:
- case VARBINARY:
- ((VarBinaryVector) vector).setSafe(rowId, rowData.getBinary(ordinal));
- return;
- case DECIMAL:
- DecimalType decimalType = (DecimalType) type;
- DecimalData decimal = rowData.getDecimal(ordinal,
decimalType.getPrecision(), decimalType.getScale());
- ((DecimalVector) vector).setSafe(rowId, decimal.toBigDecimal());
- return;
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- TimestampData timestamp = rowData.getTimestamp(ordinal,
getPrecision(type));
- long micros = timestampToMicros(timestamp, getPrecision(type),
utcTimestamp);
- ((TimeStampMicroVector) vector).setSafe(rowId, micros);
- return;
- default:
- throw unsupported(type);
- }
- }
-
private static Object readValue(LogicalType type, ValueVector vector, int
rowId) {
+ if (vector.isNull(rowId)) {
+ return null;
+ }
switch (type.getTypeRoot()) {
case BOOLEAN:
return ((BitVector) vector).get(rowId) == 1;
@@ -201,15 +152,32 @@ public final class HoodieFlinkLanceArrowUtils {
return DecimalData.fromBigDecimal(decimal, decimalType.getPrecision(),
decimalType.getScale());
case TIMESTAMP_WITHOUT_TIME_ZONE:
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- long micros = ((TimeStampMicroVector) vector).get(rowId);
+ long micros = ((TimeStampVector) vector).get(rowId);
return TimestampData.fromEpochMillis(Math.floorDiv(micros, 1000L),
(int) Math.floorMod(micros, 1000L) * 1000);
+ case ROW:
+ return readRow((RowType) type, (StructVector) vector, rowId);
+ case ARRAY:
+ return readArray((ArrayType) type, (ListVector) vector, rowId);
default:
throw unsupported(type);
}
}
private static Field toArrowField(String name, LogicalType type) {
- return new Field(name, FieldType.nullable(toArrowType(type)),
Collections.emptyList());
+ List<Field> children = new ArrayList<>();
+ switch (type.getTypeRoot()) {
+ case ROW:
+ for (RowType.RowField field : ((RowType) type).getFields()) {
+ children.add(toArrowField(field.getName(), field.getType()));
+ }
+ break;
+ case ARRAY:
+ children.add(toArrowField("element", ((ArrayType)
type).getElementType()));
+ break;
+ default:
+ break;
+ }
+ return new Field(name, new FieldType(type.isNullable(), toArrowType(type),
null), children);
}
private static ArrowType toArrowType(LogicalType type) {
@@ -245,61 +213,96 @@ public final class HoodieFlinkLanceArrowUtils {
return new ArrowType.Timestamp(TimeUnit.MICROSECOND, null);
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
return new ArrowType.Timestamp(TimeUnit.MICROSECOND, "UTC");
+ case ROW:
+ return ArrowType.Struct.INSTANCE;
+ case ARRAY:
+ return ArrowType.List.INSTANCE;
default:
throw unsupported(type);
}
}
- private static LogicalType toLogicalType(ArrowType arrowType) {
+ private static LogicalType toLogicalType(Field field, String path) {
+ ArrowType arrowType = field.getType();
+ LogicalType logicalType;
if (arrowType instanceof ArrowType.Bool) {
- return new BooleanType();
+ logicalType = new BooleanType();
} else if (arrowType instanceof ArrowType.Int) {
ArrowType.Int intType = (ArrowType.Int) arrowType;
switch (intType.getBitWidth()) {
case 8:
- return new TinyIntType();
+ logicalType = new TinyIntType();
+ break;
case 16:
- return new SmallIntType();
+ logicalType = new SmallIntType();
+ break;
case 32:
- return new IntType();
+ logicalType = new IntType();
+ break;
case 64:
- return new BigIntType();
+ logicalType = new BigIntType();
+ break;
default:
throw new HoodieNotSupportedException("Unsupported Arrow int width
for Lance Flink reader: " + intType.getBitWidth());
}
} else if (arrowType instanceof ArrowType.FloatingPoint) {
ArrowType.FloatingPoint fp = (ArrowType.FloatingPoint) arrowType;
- return fp.getPrecision() == FloatingPointPrecision.SINGLE
+ logicalType = fp.getPrecision() == FloatingPointPrecision.SINGLE
? new FloatType()
: new DoubleType();
} else if (arrowType instanceof ArrowType.Utf8) {
- return new VarCharType();
+ logicalType = new VarCharType();
} else if (arrowType instanceof ArrowType.Binary) {
- return new VarBinaryType();
+ logicalType = new VarBinaryType();
} else if (arrowType instanceof ArrowType.Date) {
- return new DateType();
+ logicalType = new DateType();
} else if (arrowType instanceof ArrowType.Time) {
- return new TimeType();
+ logicalType = new TimeType();
} else if (arrowType instanceof ArrowType.Decimal) {
ArrowType.Decimal decimal = (ArrowType.Decimal) arrowType;
- return new DecimalType(decimal.getPrecision(), decimal.getScale());
+ logicalType = new DecimalType(decimal.getPrecision(),
decimal.getScale());
} else if (arrowType instanceof ArrowType.Timestamp) {
ArrowType.Timestamp timestamp = (ArrowType.Timestamp) arrowType;
- return timestamp.getTimezone() == null
+ logicalType = timestamp.getTimezone() == null
? new TimestampType(6)
: new LocalZonedTimestampType(6);
+ } else if (arrowType instanceof ArrowType.Struct) {
+ List<RowType.RowField> fields = new
ArrayList<>(field.getChildren().size());
+ for (Field child : field.getChildren()) {
+ fields.add(new RowType.RowField(child.getName(), toLogicalType(child,
path + "." + child.getName())));
+ }
+ logicalType = new RowType(field.isNullable(), fields);
+ } else if (arrowType instanceof ArrowType.List) {
+ ValidationUtils.checkArgument(field.getChildren().size() == 1,
+ 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 {
+ throw new HoodieNotSupportedException("Unsupported Arrow type for Lance
Flink reader at '" + path + "': " + arrowType);
}
- throw new HoodieNotSupportedException("Unsupported Arrow type for Lance
Flink reader: " + arrowType);
+ return logicalType.copy(field.isNullable());
}
- private static long timestampToMicros(TimestampData timestampData, int
precision, boolean utcTimestamp) {
- long millis = utcTimestamp ? timestampData.getMillisecond() :
timestampData.toTimestamp().getTime();
- return precision > 3 && utcTimestamp
- ? millis * 1000L + timestampData.getNanoOfMillisecond() / 1000L
- : millis * 1000L;
+ private static RowData readRow(RowType rowType, StructVector vector, int
rowId) {
+ GenericRowData row = new GenericRowData(rowType.getFieldCount());
+ for (int i = 0; i < rowType.getFieldCount(); i++) {
+ row.setField(i, readValue(rowType.getTypeAt(i),
vector.getChildByOrdinal(i), rowId));
+ }
+ return row;
+ }
+
+ private static ArrayData readArray(ArrayType arrayType, ListVector vector,
int rowId) {
+ int startIndex = vector.getElementStartIndex(rowId);
+ int endIndex = vector.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);
+ }
+ return new GenericArrayData(values);
}
private static HoodieNotSupportedException unsupported(LogicalType type) {
- return new HoodieNotSupportedException("Flink Lance base-file support
currently supports primitive append-only columns; unsupported type: " + type);
+ return new HoodieNotSupportedException("Flink Lance base-file support
currently supports primitive, ROW, and ARRAY columns; unsupported type: " +
type);
}
}
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/lance/LanceRowDataWriter.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/lance/LanceRowDataWriter.java
new file mode 100644
index 000000000000..41c2481d5987
--- /dev/null
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/lance/LanceRowDataWriter.java
@@ -0,0 +1,482 @@
+/*
+ * 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.io.storage.row.lance;
+
+import org.apache.hudi.exception.HoodieNotSupportedException;
+
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.DateDayVector;
+import org.apache.arrow.vector.DecimalVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.Float4Vector;
+import org.apache.arrow.vector.Float8Vector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.SmallIntVector;
+import org.apache.arrow.vector.TimeMilliVector;
+import org.apache.arrow.vector.TimeStampVector;
+import org.apache.arrow.vector.TinyIntVector;
+import org.apache.arrow.vector.VarBinaryVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.complex.ListVector;
+import org.apache.arrow.vector.complex.StructVector;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.List;
+
+import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getPrecision;
+
+/**
+ * Writes Flink {@link RowData} values into Arrow vectors used by Lance base
files.
+ *
+ * <p>The type-specific writer tree is created once for the supplied Arrow
vectors. Record writes
+ * then use direct {@link RowData} and {@link ArrayData} accessors without
creating field or element
+ * getters on the record-level path.
+ *
+ * <p>Supports primitive types, {@code ROW}, and {@code ARRAY}. {@code MAP} is
not supported by the
+ * current Lance Java writer. Enable it once Lance adds Java writer support
for {@code MAP}.
+ *
+ * @see <a href="https://github.com/lance-format/lance/issues/3620";>Lance
issue #3620</a>
+ */
+public class LanceRowDataWriter {
+
+ private final FieldWriter[] fieldWriters;
+
+ public LanceRowDataWriter(RowType rowType, List<FieldVector> vectors,
boolean utcTimestamp) {
+ this.fieldWriters = new FieldWriter[rowType.getFieldCount()];
+ for (int i = 0; i < fieldWriters.length; i++) {
+ fieldWriters[i] = createWriter(rowType.getTypeAt(i), vectors.get(i),
utcTimestamp);
+ }
+ }
+
+ public void write(RowData row, int rowId) {
+ for (int i = 0; i < fieldWriters.length; i++) {
+ fieldWriters[i].write(row, i, rowId);
+ }
+ }
+
+ private static FieldWriter createWriter(LogicalType type, FieldVector
vector, boolean utcTimestamp) {
+ switch (type.getTypeRoot()) {
+ case BOOLEAN:
+ return new BooleanWriter((BitVector) vector);
+ case TINYINT:
+ return new TinyIntWriter((TinyIntVector) vector);
+ case SMALLINT:
+ return new SmallIntWriter((SmallIntVector) vector);
+ case INTEGER:
+ return new IntWriter((IntVector) vector);
+ case DATE:
+ return new DateWriter((DateDayVector) vector);
+ case TIME_WITHOUT_TIME_ZONE:
+ return new TimeWriter((TimeMilliVector) vector);
+ case BIGINT:
+ return new BigIntWriter((BigIntVector) vector);
+ case FLOAT:
+ return new FloatWriter((Float4Vector) vector);
+ case DOUBLE:
+ return new DoubleWriter((Float8Vector) vector);
+ case CHAR:
+ case VARCHAR:
+ return new StringWriter((VarCharVector) vector);
+ case BINARY:
+ case VARBINARY:
+ return new BinaryWriter((VarBinaryVector) vector);
+ case DECIMAL:
+ return new DecimalWriter((DecimalVector) vector, (DecimalType) type);
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return new TimestampWriter((TimeStampVector) vector,
getPrecision(type), utcTimestamp);
+ case ROW:
+ RowType rowType = (RowType) type;
+ StructVector structVector = (StructVector) vector;
+ FieldWriter[] fieldWriters = new FieldWriter[rowType.getFieldCount()];
+ for (int i = 0; i < fieldWriters.length; i++) {
+ fieldWriters[i] = createWriter(
+ rowType.getTypeAt(i), (FieldVector)
structVector.getChildByOrdinal(i), utcTimestamp);
+ }
+ return new RowWriter(structVector, fieldWriters);
+ case ARRAY:
+ ArrayType arrayType = (ArrayType) type;
+ ListVector listVector = (ListVector) vector;
+ FieldWriter elementWriter = createWriter(
+ arrayType.getElementType(), listVector.getDataVector(),
utcTimestamp);
+ return new ArrayWriter(listVector, elementWriter);
+ default:
+ throw unsupported(type);
+ }
+ }
+
+ private abstract static class FieldWriter {
+ protected final FieldVector fieldVector;
+
+ private FieldWriter(FieldVector vector) {
+ this.fieldVector = vector;
+ }
+
+ final void write(RowData row, int ordinal, int rowId) {
+ if (row.isNullAt(ordinal)) {
+ fieldVector.setNull(rowId);
+ } else {
+ writeNonNull(row, ordinal, rowId);
+ }
+ }
+
+ final void write(ArrayData array, int ordinal, int rowId) {
+ if (array.isNullAt(ordinal)) {
+ fieldVector.setNull(rowId);
+ } else {
+ writeNonNull(array, ordinal, rowId);
+ }
+ }
+
+ abstract void writeNonNull(RowData row, int ordinal, int rowId);
+
+ abstract void writeNonNull(ArrayData array, int ordinal, int rowId);
+ }
+
+ private static class BooleanWriter extends FieldWriter {
+ private final BitVector vector;
+
+ private BooleanWriter(BitVector vector) {
+ super(vector);
+ this.vector = vector;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, row.getBoolean(ordinal) ? 1 : 0);
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, array.getBoolean(ordinal) ? 1 : 0);
+ }
+ }
+
+ private static class TinyIntWriter extends FieldWriter {
+ private final TinyIntVector vector;
+
+ private TinyIntWriter(TinyIntVector vector) {
+ super(vector);
+ this.vector = vector;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, row.getByte(ordinal));
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, array.getByte(ordinal));
+ }
+ }
+
+ private static class SmallIntWriter extends FieldWriter {
+ private final SmallIntVector vector;
+
+ private SmallIntWriter(SmallIntVector vector) {
+ super(vector);
+ this.vector = vector;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, row.getShort(ordinal));
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, array.getShort(ordinal));
+ }
+ }
+
+ private static class IntWriter extends FieldWriter {
+ private final IntVector vector;
+
+ private IntWriter(IntVector vector) {
+ super(vector);
+ this.vector = vector;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, row.getInt(ordinal));
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, array.getInt(ordinal));
+ }
+ }
+
+ private static class DateWriter extends FieldWriter {
+ private final DateDayVector vector;
+
+ private DateWriter(DateDayVector vector) {
+ super(vector);
+ this.vector = vector;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, row.getInt(ordinal));
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, array.getInt(ordinal));
+ }
+ }
+
+ private static class TimeWriter extends FieldWriter {
+ private final TimeMilliVector vector;
+
+ private TimeWriter(TimeMilliVector vector) {
+ super(vector);
+ this.vector = vector;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, row.getInt(ordinal));
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, array.getInt(ordinal));
+ }
+ }
+
+ private static class BigIntWriter extends FieldWriter {
+ private final BigIntVector vector;
+
+ private BigIntWriter(BigIntVector vector) {
+ super(vector);
+ this.vector = vector;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, row.getLong(ordinal));
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, array.getLong(ordinal));
+ }
+ }
+
+ private static class FloatWriter extends FieldWriter {
+ private final Float4Vector vector;
+
+ private FloatWriter(Float4Vector vector) {
+ super(vector);
+ this.vector = vector;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, row.getFloat(ordinal));
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, array.getFloat(ordinal));
+ }
+ }
+
+ private static class DoubleWriter extends FieldWriter {
+ private final Float8Vector vector;
+
+ private DoubleWriter(Float8Vector vector) {
+ super(vector);
+ this.vector = vector;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, row.getDouble(ordinal));
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, array.getDouble(ordinal));
+ }
+ }
+
+ private static class StringWriter extends FieldWriter {
+ private final VarCharVector vector;
+
+ private StringWriter(VarCharVector vector) {
+ super(vector);
+ this.vector = vector;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, row.getString(ordinal).toBytes());
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, array.getString(ordinal).toBytes());
+ }
+ }
+
+ private static class BinaryWriter extends FieldWriter {
+ private final VarBinaryVector vector;
+
+ private BinaryWriter(VarBinaryVector vector) {
+ super(vector);
+ this.vector = vector;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, row.getBinary(ordinal));
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, array.getBinary(ordinal));
+ }
+ }
+
+ private static class DecimalWriter extends FieldWriter {
+ private final DecimalVector vector;
+ private final int precision;
+ private final int scale;
+
+ private DecimalWriter(DecimalVector vector, DecimalType decimalType) {
+ super(vector);
+ this.vector = vector;
+ this.precision = decimalType.getPrecision();
+ this.scale = decimalType.getScale();
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, row.getDecimal(ordinal, precision,
scale).toBigDecimal());
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, array.getDecimal(ordinal, precision,
scale).toBigDecimal());
+ }
+ }
+
+ private static class TimestampWriter extends FieldWriter {
+ private final TimeStampVector vector;
+ private final int precision;
+ private final boolean utcTimestamp;
+
+ private TimestampWriter(TimeStampVector vector, int precision, boolean
utcTimestamp) {
+ super(vector);
+ this.vector = vector;
+ this.precision = precision;
+ this.utcTimestamp = utcTimestamp;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ vector.setSafe(rowId, timestampToMicros(row.getTimestamp(ordinal,
precision), precision, utcTimestamp));
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ vector.setSafe(rowId, timestampToMicros(array.getTimestamp(ordinal,
precision), precision, utcTimestamp));
+ }
+ }
+
+ private static class RowWriter extends FieldWriter {
+ private final StructVector vector;
+ private final FieldWriter[] fieldWriters;
+
+ private RowWriter(StructVector vector, FieldWriter[] fieldWriters) {
+ super(vector);
+ this.vector = vector;
+ this.fieldWriters = fieldWriters;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ writeRow(row.getRow(ordinal, fieldWriters.length), rowId);
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ writeRow(array.getRow(ordinal, fieldWriters.length), rowId);
+ }
+
+ private void writeRow(RowData row, int rowId) {
+ vector.setIndexDefined(rowId);
+ for (int i = 0; i < fieldWriters.length; i++) {
+ fieldWriters[i].write(row, i, rowId);
+ }
+ }
+ }
+
+ private static class ArrayWriter extends FieldWriter {
+ private final ListVector vector;
+ private final FieldWriter elementWriter;
+
+ private ArrayWriter(ListVector vector, FieldWriter elementWriter) {
+ super(vector);
+ this.vector = vector;
+ this.elementWriter = elementWriter;
+ }
+
+ @Override
+ void writeNonNull(RowData row, int ordinal, int rowId) {
+ writeArray(row.getArray(ordinal), rowId);
+ }
+
+ @Override
+ void writeNonNull(ArrayData array, int ordinal, int rowId) {
+ writeArray(array.getArray(ordinal), rowId);
+ }
+
+ private void writeArray(ArrayData array, int rowId) {
+ int startIndex = vector.startNewValue(rowId);
+ for (int i = 0; i < array.size(); i++) {
+ elementWriter.write(array, i, startIndex + i);
+ }
+ vector.endValue(rowId, array.size());
+ }
+ }
+
+ private static long timestampToMicros(TimestampData timestampData, int
precision, boolean utcTimestamp) {
+ long millis = utcTimestamp ? timestampData.getMillisecond() :
timestampData.toTimestamp().getTime();
+ return precision > 3 && utcTimestamp
+ ? millis * 1000L + timestampData.getNanoOfMillisecond() / 1000L
+ : millis * 1000L;
+ }
+
+ private static HoodieNotSupportedException unsupported(LogicalType type) {
+ return new HoodieNotSupportedException(
+ "Flink Lance base-file support currently supports primitive, ROW, and
ARRAY columns; unsupported type: " + type);
+ }
+}
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieFlinkLanceArrowUtils.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieFlinkLanceArrowUtils.java
index 2a9ecc5b0145..007148a4df97 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieFlinkLanceArrowUtils.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieFlinkLanceArrowUtils.java
@@ -18,25 +18,33 @@
package org.apache.hudi.io.storage.row;
+import org.apache.hudi.exception.HoodieNotSupportedException;
+import org.apache.hudi.io.storage.row.lance.HoodieFlinkLanceArrowUtils;
+
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.TimeStampMicroVector;
import org.apache.arrow.vector.types.TimeUnit;
import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.arrow.vector.types.pojo.FieldType;
-import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MapType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.TimestampType;
+import org.apache.flink.table.types.logical.VarCharType;
import org.junit.jupiter.api.Test;
import java.util.Collections;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Tests for {@link HoodieFlinkLanceArrowUtils}.
@@ -56,24 +64,6 @@ public class TestHoodieFlinkLanceArrowUtils {
assertInstanceOf(LocalZonedTimestampType.class, roundTripped.getTypeAt(1));
}
- @Test
- public void testTimestampWriteHonorsUtcTimestampFlag() {
- TimestampData timestampData = TimestampData.fromEpochMillis(1234L, 567000);
- GenericRowData rowData = GenericRowData.of(timestampData);
-
- try (BufferAllocator allocator = new RootAllocator();
- TimeStampMicroVector vector = new TimeStampMicroVector(
- "ts",
- FieldType.nullable(new ArrowType.Timestamp(TimeUnit.MICROSECOND,
null)),
- allocator)) {
- HoodieFlinkLanceArrowUtils.writeValue(new TimestampType(6), vector, 0,
rowData, 0, true);
- assertEquals(1234567L, vector.get(0));
-
- HoodieFlinkLanceArrowUtils.writeValue(new TimestampType(6), vector, 1,
rowData, 0, false);
- assertEquals(timestampData.toTimestamp().getTime() * 1000L,
vector.get(1));
- }
- }
-
@Test
public void testTimestampReadNormalizesPreEpochMicros() {
try (BufferAllocator allocator = new RootAllocator();
@@ -92,4 +82,34 @@ public class TestHoodieFlinkLanceArrowUtils {
assertEquals(TimestampData.fromEpochMillis(-1235L, 433000),
rowData.getTimestamp(0, 6));
}
}
+
+ @Test
+ public void testNestedSchemaRoundTrip() {
+ RowType rowType = nestedRowType();
+
+ RowType roundTripped = HoodieFlinkLanceArrowUtils.toRowType(
+ HoodieFlinkLanceArrowUtils.toArrowSchema(rowType));
+
+ assertEquals(rowType, roundTripped);
+ }
+
+ @Test
+ public void testRejectsMapTypeWhenWritingSchema() {
+ HoodieNotSupportedException exception =
assertThrows(HoodieNotSupportedException.class,
+ () -> HoodieFlinkLanceArrowUtils.toArrowSchema(RowType.of(
+ new LogicalType[] {new MapType(new VarCharType(), new IntType())},
+ new String[] {"attributes"})));
+ assertTrue(exception.getMessage().contains("Flink Lance base-file support
currently supports primitive, ROW, and ARRAY columns;"));
+ }
+
+ private static RowType nestedRowType() {
+ RowType profileType = RowType.of(
+ new LogicalType[] {new VarCharType(), new IntType()},
+ new String[] {"name", "age"});
+ ArrayType numbersType = new ArrayType(new IntType());
+ ArrayType profilesType = new ArrayType(profileType);
+ return RowType.of(
+ new LogicalType[] {profileType, numbersType, profilesType},
+ new String[] {"profile", "numbers", "profiles"});
+ }
}
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestLanceRowDataWriter.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestLanceRowDataWriter.java
new file mode 100644
index 000000000000..d43c2ad6d9b1
--- /dev/null
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestLanceRowDataWriter.java
@@ -0,0 +1,274 @@
+/*
+ * 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.io.storage.row;
+
+import org.apache.hudi.exception.HoodieNotSupportedException;
+import org.apache.hudi.io.storage.row.lance.HoodieFlinkLanceArrowUtils;
+import org.apache.hudi.io.storage.row.lance.LanceRowDataWriter;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.TimeStampMicroVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.ListVector;
+import org.apache.arrow.vector.complex.StructVector;
+import org.apache.arrow.vector.types.TimeUnit;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.flink.table.data.DecimalData;
+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.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.BinaryType;
+import org.apache.flink.table.types.logical.BooleanType;
+import org.apache.flink.table.types.logical.CharType;
+import org.apache.flink.table.types.logical.DateType;
+import org.apache.flink.table.types.logical.DecimalType;
+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.LocalZonedTimestampType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.SmallIntType;
+import org.apache.flink.table.types.logical.TimeType;
+import org.apache.flink.table.types.logical.TimestampType;
+import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.VarBinaryType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link LanceRowDataWriter}.
+ */
+public class TestLanceRowDataWriter {
+
+ @Test
+ public void testPrimitiveValuesAndNulls() {
+ RowType rowType = primitiveRowType();
+ TimestampData timestamp = TimestampData.fromEpochMillis(1234L, 567000);
+ TimestampData localTimestamp = TimestampData.fromEpochMillis(5678L,
123000);
+ DecimalData decimal = DecimalData.fromBigDecimal(new
BigDecimal("12345.67"), 10, 2);
+ GenericRowData values = GenericRowData.of(
+ true,
+ (byte) 12,
+ (short) 1234,
+ 123456,
+ 20000,
+ 12345678,
+ 1234567890123L,
+ 1.25F,
+ 2.5D,
+ StringData.fromString("char"),
+ StringData.fromString("varchar"),
+ new byte[] {1, 2, 3},
+ new byte[] {4, 5, 6, 7},
+ decimal,
+ timestamp,
+ localTimestamp);
+ GenericRowData nulls = new GenericRowData(rowType.getFieldCount());
+
+ try (BufferAllocator allocator = new RootAllocator();
+ VectorSchemaRoot root = VectorSchemaRoot.create(
+ HoodieFlinkLanceArrowUtils.toArrowSchema(rowType), allocator)) {
+ root.allocateNew();
+ LanceRowDataWriter writer = new LanceRowDataWriter(rowType,
root.getFieldVectors(), true);
+ writer.write(values, 0);
+ writer.write(nulls, 1);
+ root.getFieldVectors().forEach(vector -> vector.setValueCount(2));
+ root.setRowCount(2);
+
+ RowData actual = HoodieFlinkLanceArrowUtils.toRowData(rowType,
root.getFieldVectors(), 0);
+ assertEquals(true, actual.getBoolean(0));
+ assertEquals((byte) 12, actual.getByte(1));
+ assertEquals((short) 1234, actual.getShort(2));
+ assertEquals(123456, actual.getInt(3));
+ assertEquals(20000, actual.getInt(4));
+ assertEquals(12345678, actual.getInt(5));
+ assertEquals(1234567890123L, actual.getLong(6));
+ assertEquals(1.25F, actual.getFloat(7));
+ assertEquals(2.5D, actual.getDouble(8));
+ assertEquals(StringData.fromString("char"), actual.getString(9));
+ assertEquals(StringData.fromString("varchar"), actual.getString(10));
+ assertArrayEquals(new byte[] {1, 2, 3}, actual.getBinary(11));
+ assertArrayEquals(new byte[] {4, 5, 6, 7}, actual.getBinary(12));
+ assertEquals(decimal, actual.getDecimal(13, 10, 2));
+ assertEquals(timestamp, actual.getTimestamp(14, 6));
+ assertEquals(localTimestamp, actual.getTimestamp(15, 6));
+
+ RowData actualNulls = HoodieFlinkLanceArrowUtils.toRowData(rowType,
root.getFieldVectors(), 1);
+ for (int i = 0; i < rowType.getFieldCount(); i++) {
+ assertTrue(actualNulls.isNullAt(i));
+ }
+ }
+ }
+
+ @Test
+ public void testTimestampWriteHonorsUtcTimestampFlag() {
+ TimestampData timestampData = TimestampData.fromEpochMillis(1234L, 567000);
+ GenericRowData rowData = GenericRowData.of(timestampData);
+
+ try (BufferAllocator allocator = new RootAllocator();
+ TimeStampMicroVector vector = new TimeStampMicroVector(
+ "ts",
+ FieldType.nullable(new ArrowType.Timestamp(TimeUnit.MICROSECOND,
null)),
+ allocator)) {
+ RowType rowType = RowType.of(new LogicalType[] {new TimestampType(6)},
new String[] {"ts"});
+ new LanceRowDataWriter(rowType, Collections.singletonList(vector),
true).write(rowData, 0);
+ assertEquals(1234567L, vector.get(0));
+
+ new LanceRowDataWriter(rowType, Collections.singletonList(vector),
false).write(rowData, 1);
+ assertEquals(timestampData.toTimestamp().getTime() * 1000L,
vector.get(1));
+ }
+ }
+
+ @Test
+ public void testNestedValueRoundTripAndListValueCount() {
+ RowType rowType = nestedRowType();
+ GenericRowData first = GenericRowData.of(
+ GenericRowData.of(
+ StringData.fromString("alice"),
+ null,
+ new GenericArrayData(new Object[]
{StringData.fromString("primary"), null})),
+ new GenericArrayData(new Object[] {1, null}),
+ new GenericArrayData(new Object[] {
+ GenericRowData.of(
+ StringData.fromString("child"),
+ 3,
+ new GenericArrayData(new Object[0])),
+ null}),
+ new GenericArrayData(new Object[] {
+ new GenericArrayData(new Object[] {1, null}),
+ null,
+ new GenericArrayData(new Object[0])}));
+ GenericRowData second = GenericRowData.of(
+ null,
+ new GenericArrayData(new Object[0]),
+ new GenericArrayData(new Object[0]),
+ new GenericArrayData(new Object[0]));
+ GenericRowData third = new GenericRowData(rowType.getFieldCount());
+
+ try (BufferAllocator allocator = new RootAllocator();
+ VectorSchemaRoot root = VectorSchemaRoot.create(
+ HoodieFlinkLanceArrowUtils.toArrowSchema(rowType), allocator)) {
+ root.allocateNew();
+ LanceRowDataWriter writer = new LanceRowDataWriter(rowType,
root.getFieldVectors(), true);
+ writer.write(first, 0);
+ writer.write(second, 1);
+ writer.write(third, 2);
+ root.getFieldVectors().forEach(vector -> vector.setValueCount(3));
+ root.setRowCount(3);
+
+ assertEquals(first, HoodieFlinkLanceArrowUtils.toRowData(rowType,
root.getFieldVectors(), 0));
+ assertEquals(second, HoodieFlinkLanceArrowUtils.toRowData(rowType,
root.getFieldVectors(), 1));
+ assertEquals(third, HoodieFlinkLanceArrowUtils.toRowData(rowType,
root.getFieldVectors(), 2));
+ assertEquals(2, ((ListVector)
root.getVector(1)).getDataVector().getValueCount());
+ assertEquals(2, ((ListVector) ((StructVector)
root.getVector(0)).getChild("tags"))
+ .getDataVector().getValueCount());
+
+ ListVector matrixVector = (ListVector) root.getVector(3);
+ assertEquals(3, matrixVector.getDataVector().getValueCount());
+ assertEquals(2, ((ListVector)
matrixVector.getDataVector()).getDataVector().getValueCount());
+ }
+ }
+
+ @Test
+ public void testRejectsMapType() {
+ MapType mapType = new MapType(new VarCharType(), new IntType());
+
+ try (BufferAllocator allocator = new RootAllocator();
+ IntVector vector = new IntVector("attributes", allocator)) {
+ HoodieNotSupportedException exception =
assertThrows(HoodieNotSupportedException.class,
+ () -> new LanceRowDataWriter(
+ RowType.of(new LogicalType[] {mapType}, new String[]
{"attributes"}),
+ Collections.singletonList(vector),
+ true));
+ assertTrue(exception.getMessage().contains(mapUnsupportedMessage()));
+ }
+ }
+
+ private static String mapUnsupportedMessage() {
+ return "Flink Lance base-file support currently supports primitive, ROW,
and ARRAY columns;";
+ }
+
+ private static RowType nestedRowType() {
+ ArrayType tagsType = new ArrayType(new VarCharType());
+ RowType profileType = RowType.of(
+ new LogicalType[] {new VarCharType(), new IntType(), tagsType},
+ new String[] {"name", "age", "tags"});
+ ArrayType numbersType = new ArrayType(new IntType());
+ ArrayType profilesType = new ArrayType(profileType);
+ ArrayType matrixType = new ArrayType(numbersType);
+ return RowType.of(
+ new LogicalType[] {profileType, numbersType, profilesType, matrixType},
+ new String[] {"profile", "numbers", "profiles", "matrix"});
+ }
+
+ private static RowType primitiveRowType() {
+ return RowType.of(
+ new LogicalType[] {
+ new BooleanType(),
+ new TinyIntType(),
+ new SmallIntType(),
+ new IntType(),
+ new DateType(),
+ new TimeType(),
+ new BigIntType(),
+ new FloatType(),
+ new DoubleType(),
+ new CharType(4),
+ new VarCharType(),
+ new BinaryType(3),
+ new VarBinaryType(),
+ new DecimalType(10, 2),
+ new TimestampType(6),
+ new LocalZonedTimestampType(6)},
+ new String[] {
+ "boolean_value",
+ "tinyint_value",
+ "smallint_value",
+ "int_value",
+ "date_value",
+ "time_value",
+ "bigint_value",
+ "float_value",
+ "double_value",
+ "char_value",
+ "varchar_value",
+ "binary_value",
+ "varbinary_value",
+ "decimal_value",
+ "timestamp_value",
+ "local_timestamp_value"});
+ }
+}
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 19a96d7c8f36..02d1f3c073f3 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
@@ -34,7 +34,7 @@ import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieValidationException;
import org.apache.hudi.io.memory.HoodieArrowAllocator;
-import org.apache.hudi.io.storage.row.HoodieFlinkLanceArrowUtils;
+import org.apache.hudi.io.storage.row.lance.HoodieFlinkLanceArrowUtils;
import org.apache.hudi.source.ExpressionPredicates;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.util.HoodieSchemaConverter;
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
index 4d03c3d9cc82..07c3f0c05ecd 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
@@ -1734,6 +1734,49 @@ public class ITTestHoodieDataSource {
assertRowsEquals(projectedRows, "[+I[Alice, id1], +I[Bob, id2]]");
}
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testLanceFormatNestedTypesUpsertWriteAndRead(HoodieTableType tableType)
{
+ String createHoodieTable = sql("lance_nested")
+ .field("id int not null")
+ .field("ts bigint")
+ .field("f_row row(f_name varchar(10), f_age int)")
+ .field("f_array array<row(f_name varchar(10), f_age int)>")
+ .field("f_nested_array array<row(f_scores array<int>)>")
+ .pkField("id")
+ .noPartition()
+ .option(FlinkOptions.PATH, tempFile.getAbsolutePath())
+ .option(FlinkOptions.OPERATION, "upsert")
+ .option(FlinkOptions.ORDERING_FIELDS, "ts")
+ .option(FlinkOptions.TABLE_TYPE, tableType)
+ .option("hoodie.table.base.file.format", "LANCE")
+ .end();
+ batchTableEnv.executeSql(createHoodieTable);
+
+ execInsertSql(batchTableEnv, "insert into lance_nested values "
+ + "(1, 1, ROW('alice', 30), ARRAY[ROW('child1', 1), ROW('child2', 2)],
"
+ + "ARRAY[ROW(ARRAY[1, 2]), ROW(ARRAY[3])]),"
+ + "(2, 2, ROW('bob', 31), ARRAY[ROW('child3', 3)],
ARRAY[ROW(ARRAY[4])])");
+
+ execInsertSql(batchTableEnv, "insert into lance_nested values "
+ + "(1, 3, ROW('alice_v2', 32), ARRAY[ROW('child4', 4)],
ARRAY[ROW(ARRAY[5, 6])]),"
+ + "(3, 4, ROW('charlie', 33), ARRAY[ROW('child5', 5)],
ARRAY[ROW(ARRAY[7])])");
+
+ List<Row> rows = CollectionUtil.iterableToList(
+ () -> batchTableEnv.sqlQuery("select * from
lance_nested").execute().collect());
+ assertRowsEqualsUnordered(Arrays.asList(
+ row(1, 3L, row("alice_v2", 32), array(row("child4", 4)),
array(row((Object) array(5, 6)))),
+ row(2, 2L, row("bob", 31), array(row("child3", 3)), array(row((Object)
array(4)))),
+ row(3, 4L, row("charlie", 33), array(row("child5", 5)),
array(row((Object) array(7))))), rows);
+
+ List<Row> projectedRows = CollectionUtil.iterableToList(
+ () -> batchTableEnv.sqlQuery("select f_nested_array, id from
lance_nested").execute().collect());
+ assertRowsEqualsUnordered(Arrays.asList(
+ row(array(row((Object) array(5, 6))), 1),
+ row(array(row((Object) array(4))), 2),
+ row(array(row((Object) array(7))), 3)), projectedRows);
+ }
+
@Test
void testLanceFormatCopyOnWriteUpsertWriteAndRead() {
String createHoodieTable = sql("t1")
