hudi-agent commented on code in PR #19242:
URL: https://github.com/apache/hudi/pull/19242#discussion_r3565809373
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieFlinkLanceArrowUtils.java:
##########
@@ -245,51 +286,122 @@ private static ArrowType toArrowType(LogicalType type) {
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) {
+ Field element = onlyChild(field, path, "list");
+ logicalType = new ArrayType(field.isNullable(), toLogicalType(element,
path + "[]"));
+ } else {
+ throw new HoodieNotSupportedException("Unsupported Arrow type for Lance
Flink reader at '" + path + "': " + arrowType);
+ }
+ return logicalType.copy(field.isNullable());
+ }
+
+ private static void writeRow(RowType rowType, StructVector vector, int
rowId, RowData row, boolean utcTimestamp) {
+ vector.setIndexDefined(rowId);
+ for (int i = 0; i < rowType.getFieldCount(); i++) {
+ LogicalType childType = rowType.getTypeAt(i);
+ Object childValue = RowData.createFieldGetter(childType,
i).getFieldOrNull(row);
+ writeValue(childType, (FieldVector) vector.getChildByOrdinal(i), rowId,
childValue, utcTimestamp);
}
- throw new HoodieNotSupportedException("Unsupported Arrow type for Lance
Flink reader: " + arrowType);
+ }
+
+ private static void writeArray(ArrayType arrayType, ListVector vector, int
rowId, ArrayData array, boolean utcTimestamp) {
+ int startIndex = vector.startNewValue(rowId);
+ LogicalType elementType = arrayType.getElementType();
+ ArrayData.ElementGetter elementGetter =
ArrayData.createElementGetter(elementType);
+ FieldVector dataVector = vector.getDataVector();
+ for (int i = 0; i < array.size(); i++) {
+ writeValue(elementType, dataVector, startIndex + i,
elementGetter.getElementOrNull(array, i), utcTimestamp);
+ }
+ vector.endValue(rowId, array.size());
+ }
+
+ 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 Field onlyChild(Field field, String path, String type) {
+ if (field.getChildren().size() != 1) {
+ throw invalidArrowSchema(path, type + " must contain exactly one child
field");
+ }
+ return field.getChildren().get(0);
+ }
Review Comment:
🤖 nit: `type` as a `String` parameter name is a bit confusing here since
every other `type` in this file refers to a `LogicalType` or `ArrowType` object
— could you rename it to `typeName` (or `arrowKind`) to make it clear this is
just a label for the error message?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
