This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 36ce7c97ee [flink] Preserve vector layouts in schema-aware row
conversion (#10677)
36ce7c97ee is described below
commit 36ce7c97ee2090d7cfc72c1f7e055cca892f12ba
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 20 12:27:20 2026 +0800
[flink] Preserve vector layouts in schema-aware row conversion (#10677)
---
.../java/org/apache/paimon/flink/FlinkRowData.java | 76 +++++++++++++++++++---
.../flink/compact/IncrementalClusterCompact.java | 2 +-
.../org/apache/paimon/flink/sorter/SortUtils.java | 2 +-
.../paimon/flink/FlinkRowDataWithVectorTest.java | 75 +++++++++++++++++++++
4 files changed, 144 insertions(+), 11 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowData.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowData.java
index 2016ab4bf4..8db5c7d9e0 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowData.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowData.java
@@ -24,6 +24,12 @@ import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalMap;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.MultisetType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VectorType;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.DecimalData;
@@ -36,15 +42,23 @@ import org.apache.flink.types.RowKind;
import org.apache.flink.types.variant.BinaryVariant;
import org.apache.flink.types.variant.Variant;
+import javax.annotation.Nullable;
+
import static org.apache.paimon.flink.FlinkRowWrapper.fromFlinkRowKind;
/** Convert to Flink row data. */
public class FlinkRowData implements RowData {
protected InternalRow row;
+ @Nullable private final RowType rowType;
public FlinkRowData(InternalRow row) {
+ this(row, null);
+ }
+
+ public FlinkRowData(InternalRow row, @Nullable RowType rowType) {
this.row = row;
+ this.rowType = rowType;
}
public FlinkRowData replace(InternalRow row) {
@@ -134,17 +148,39 @@ public class FlinkRowData implements RowData {
@Override
public ArrayData getArray(int pos) {
- return new FlinkArrayData(row.getArray(pos));
+ // VECTOR and ARRAY have different binary layouts even though Flink
exposes both as ARRAY.
+ DataType type = fieldType(pos);
+ return new FlinkArrayData(
+ type instanceof VectorType ? row.getVector(pos) :
row.getArray(pos),
+ arrayElementType(type));
}
@Override
public MapData getMap(int pos) {
- return new FlinkMapData(row.getMap(pos));
+ return new FlinkMapData(row.getMap(pos), fieldType(pos));
}
@Override
public RowData getRow(int pos, int numFields) {
- return new FlinkRowData(row.getRow(pos, numFields));
+ DataType type = fieldType(pos);
+ return new FlinkRowData(
+ row.getRow(pos, numFields), type instanceof RowType ?
(RowType) type : null);
+ }
+
+ @Nullable
+ private DataType fieldType(int pos) {
+ return rowType == null ? null : rowType.getTypeAt(pos);
+ }
+
+ @Nullable
+ private static DataType arrayElementType(@Nullable DataType type) {
+ if (type instanceof ArrayType) {
+ return ((ArrayType) type).getElementType();
+ }
+ if (type instanceof VectorType) {
+ return ((VectorType) type).getElementType();
+ }
+ return null;
}
public Variant getVariant(int pos) {
@@ -156,9 +192,15 @@ public class FlinkRowData implements RowData {
protected static class FlinkArrayData implements ArrayData {
private final InternalArray array;
+ @Nullable private final DataType elementType;
protected FlinkArrayData(InternalArray array) {
+ this(array, null);
+ }
+
+ private FlinkArrayData(InternalArray array, @Nullable DataType
elementType) {
this.array = array;
+ this.elementType = elementType;
}
@Override
@@ -238,17 +280,21 @@ public class FlinkRowData implements RowData {
@Override
public ArrayData getArray(int pos) {
- return new FlinkArrayData(array.getArray(pos));
+ return new FlinkArrayData(
+ elementType instanceof VectorType ? array.getVector(pos) :
array.getArray(pos),
+ arrayElementType(elementType));
}
@Override
public MapData getMap(int pos) {
- return new FlinkMapData(array.getMap(pos));
+ return new FlinkMapData(array.getMap(pos), elementType);
}
@Override
public RowData getRow(int pos, int numFields) {
- return new FlinkRowData(array.getRow(pos, numFields));
+ return new FlinkRowData(
+ array.getRow(pos, numFields),
+ elementType instanceof RowType ? (RowType) elementType :
null);
}
@Override
@@ -290,9 +336,21 @@ public class FlinkRowData implements RowData {
private static class FlinkMapData implements MapData {
private final InternalMap map;
+ @Nullable private final DataType keyType;
+ @Nullable private final DataType valueType;
- private FlinkMapData(InternalMap map) {
+ private FlinkMapData(InternalMap map, @Nullable DataType type) {
this.map = map;
+ if (type instanceof MapType) {
+ this.keyType = ((MapType) type).getKeyType();
+ this.valueType = ((MapType) type).getValueType();
+ } else {
+ this.keyType =
+ type instanceof MultisetType
+ ? ((MultisetType) type).getElementType()
+ : null;
+ this.valueType = null;
+ }
}
@Override
@@ -302,12 +360,12 @@ public class FlinkRowData implements RowData {
@Override
public ArrayData keyArray() {
- return new FlinkArrayData(map.keyArray());
+ return new FlinkArrayData(map.keyArray(), keyType);
}
@Override
public ArrayData valueArray() {
- return new FlinkArrayData(map.valueArray());
+ return new FlinkArrayData(map.valueArray(), valueType);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/IncrementalClusterCompact.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/IncrementalClusterCompact.java
index 25d43ddbc6..a4f25486e1 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/IncrementalClusterCompact.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/IncrementalClusterCompact.java
@@ -319,7 +319,7 @@ public class IncrementalClusterCompact {
},
InternalTypeInfo.fromRowType(rowType))
.setParallelism(sinkParallelism)
- .map(FlinkRowData::new, input.getType())
+ .map(row -> new FlinkRowData(row, rowType), input.getType())
.setParallelism(sinkParallelism);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
index c841afa3f5..adfb5c60d5 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
@@ -201,7 +201,7 @@ public class SortUtils {
},
InternalTypeInfo.fromRowType(valueRowType))
.setParallelism(sinkParallelism)
- .map(FlinkRowData::new, inputStream.getType())
+ .map(row -> new FlinkRowData(row, valueRowType),
inputStream.getType())
.setParallelism(sinkParallelism);
} else {
return rangeShuffleResult
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRowDataWithVectorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRowDataWithVectorTest.java
index ba8542d395..63317299bd 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRowDataWithVectorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRowDataWithVectorTest.java
@@ -18,13 +18,23 @@
package org.apache.paimon.flink;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryVector;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericMap;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.KeyProjectedRow;
import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.MapData;
import org.junit.jupiter.api.Test;
+import java.util.Collections;
+
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link FlinkRowData} and vector type. */
@@ -41,6 +51,71 @@ public class FlinkRowDataWithVectorTest {
assertThat(arrayData.toFloatArray()).isEqualTo(values);
}
+ @Test
+ public void testBinaryVectorAndArrayAsArrayData() {
+ float[] vector = new float[] {0.0f, 1.0f, 2.0f};
+ float[] array = new float[] {3.0f, 4.0f};
+ RowType rowType =
+ DataTypes.ROW(
+ DataTypes.VECTOR(3, DataTypes.FLOAT()),
DataTypes.ARRAY(DataTypes.FLOAT()));
+ BinaryRow row =
+ new InternalRowSerializer(rowType)
+ .toBinaryRow(
+ GenericRow.of(
+
BinaryVector.fromPrimitiveArray(vector),
+ new GenericArray(array)));
+
+ FlinkRowData rowData = new FlinkRowData(row, rowType);
+
+ assertThat(rowData.getArray(0).toFloatArray()).isEqualTo(vector);
+ assertThat(rowData.getArray(1).toFloatArray()).isEqualTo(array);
+ }
+
+ @Test
+ public void testProjectedBinaryVectorAsArrayData() {
+ float[] vector = new float[] {0.0f, 1.0f, 2.0f};
+ RowType valueType = DataTypes.ROW(DataTypes.VECTOR(3,
DataTypes.FLOAT()));
+ BinaryRow sortedRow =
+ new InternalRowSerializer(DataTypes.INT(),
valueType.getTypeAt(0))
+ .toBinaryRow(GenericRow.of(5,
BinaryVector.fromPrimitiveArray(vector)));
+ KeyProjectedRow projected = new KeyProjectedRow(new int[]
{1}).replaceRow(sortedRow);
+
+ assertThat(new FlinkRowData(projected,
valueType).getArray(0).toFloatArray())
+ .isEqualTo(vector);
+ }
+
+ @Test
+ public void testNestedBinaryVectorsAsArrayData() {
+ float[] values = new float[] {0.0f, 1.0f, 2.0f};
+ BinaryVector vector = BinaryVector.fromPrimitiveArray(values);
+ RowType nestedRowType = DataTypes.ROW(DataTypes.VECTOR(3,
DataTypes.FLOAT()));
+ RowType rowType =
+ DataTypes.ROW(
+ nestedRowType,
+ DataTypes.ARRAY(nestedRowType),
+ DataTypes.MAP(
+ nestedRowType.getTypeAt(0),
+ DataTypes.ARRAY(nestedRowType.getTypeAt(0))));
+ BinaryRow row =
+ new InternalRowSerializer(rowType)
+ .toBinaryRow(
+ GenericRow.of(
+ GenericRow.of(vector),
+ new GenericArray(new Object[]
{GenericRow.of(vector)}),
+ new GenericMap(
+ Collections.singletonMap(
+ vector,
+ new GenericArray(new
Object[] {vector})))));
+
+ FlinkRowData rowData = new FlinkRowData(row, rowType);
+
+ assertThat(rowData.getRow(0,
1).getArray(0).toFloatArray()).isEqualTo(values);
+ assertThat(rowData.getArray(1).getRow(0,
1).getArray(0).toFloatArray()).isEqualTo(values);
+ MapData map = rowData.getMap(2);
+
assertThat(map.keyArray().getArray(0).toFloatArray()).isEqualTo(values);
+
assertThat(map.valueArray().getArray(0).getArray(0).toFloatArray()).isEqualTo(values);
+ }
+
@Test
public void testNullVector() {
InternalRow row = GenericRow.of(1, null);
