This is an automated email from the ASF dual-hosted git repository.
KevinyhZou pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 0dbb30fb21 [GLUTEN-12459][FLINK] feat: supprt RowKind via $row_kind
TINYINT column between Flink and Velox (#12510)
0dbb30fb21 is described below
commit 0dbb30fb2178273bff3c0b739246de574c8a0fef
Author: GGboom <[email protected]>
AuthorDate: Tue Sep 1 16:14:27 2026 +0800
[GLUTEN-12459][FLINK] feat: supprt RowKind via $row_kind TINYINT column
between Flink and Velox (#12510)
* Carry RowKind via $row_kind TINYINT column between Flink and Velox
---
.github/workflows/flink.yml | 5 +-
gluten-flink/docs/Flink.md | 4 +-
.../gluten/vectorized/ArrowVectorAccessor.java | 13 ++
.../gluten/vectorized/ArrowVectorWriter.java | 26 ++++
.../vectorized/FlinkRowToVLVectorConvertor.java | 47 +++++-
.../FlinkRowToVLVectorConvertorTest.java | 173 +++++++++++++++++++++
6 files changed, 261 insertions(+), 7 deletions(-)
diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml
index ef8c36c311..37a6acbf88 100644
--- a/.github/workflows/flink.yml
+++ b/.github/workflows/flink.yml
@@ -87,8 +87,9 @@ jobs:
export VELOX_DEPENDENCY_SOURCE=BUNDLED
export fmt_SOURCE=BUNDLED
export folly_SOURCE=BUNDLED
- git clone -b gluten-0530 https://github.com/bigo-sg/velox4j.git
- cd velox4j && git reset --hard
42a77de002707b6811791ba2d728f169aeb5d7bb
+ git clone -b gluten-20260829 https://github.com/bigo-sg/velox4j.git
+ cd velox4j
+ git reset --hard 26c7715278e6f6795084f6334998cb2ce382f7aa
git apply $GITHUB_WORKSPACE/gluten-flink/patches/fix-velox4j.patch
$GITHUB_WORKSPACE/build/mvn clean install -DskipTests -Dgpg.skip
-Dspotless.skip=true
cd ..
diff --git a/gluten-flink/docs/Flink.md b/gluten-flink/docs/Flink.md
index 3a4fcf9fdb..b5586eea22 100644
--- a/gluten-flink/docs/Flink.md
+++ b/gluten-flink/docs/Flink.md
@@ -46,9 +46,9 @@ As some features have not been committed to upstream, you
have to use the follow
```bash
## fetch velox4j code
-git clone -b gluten-0530 https://github.com/bigo-sg/velox4j.git
+git clone -b gluten-20260829 https://github.com/bigo-sg/velox4j.git
cd velox4j
-git reset --hard 42a77de002707b6811791ba2d728f169aeb5d7bb
+git reset --hard 26c7715278e6f6795084f6334998cb2ce382f7aa
mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true
```
**Get gluten**
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorAccessor.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorAccessor.java
index c397e6fada..41ec500f38 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorAccessor.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorAccessor.java
@@ -31,6 +31,7 @@ import org.apache.arrow.vector.FieldVector;
import org.apache.arrow.vector.Float8Vector;
import org.apache.arrow.vector.IntVector;
import org.apache.arrow.vector.TimeStampMicroVector;
+import org.apache.arrow.vector.TinyIntVector;
import org.apache.arrow.vector.VarCharVector;
import org.apache.arrow.vector.complex.ListVector;
import org.apache.arrow.vector.complex.MapVector;
@@ -56,6 +57,7 @@ public abstract class ArrowVectorAccessor {
Map.ofEntries(
Map.entry(BitVector.class, vector -> new
BooleanVectorAccessor(vector)),
Map.entry(IntVector.class, vector -> new IntVectorAccessor(vector)),
+ Map.entry(TinyIntVector.class, vector -> new
TinyIntVectorAccessor(vector)),
Map.entry(BigIntVector.class, vector -> new
BigIntVectorAccessor(vector)),
Map.entry(Float8Vector.class, vector -> new
DoubleVectorAccessor(vector)),
Map.entry(DecimalVector.class, vector -> new
DecimalVectorAccessor(vector)),
@@ -128,6 +130,17 @@ class IntVectorAccessor extends
BaseArrowVectorAccessor<IntVector> {
}
}
+class TinyIntVectorAccessor extends BaseArrowVectorAccessor<TinyIntVector> {
+ public TinyIntVectorAccessor(FieldVector vector) {
+ super(vector);
+ }
+
+ @Override
+ protected Object getImpl(int rowIndex) {
+ return typedVector.get(rowIndex);
+ }
+}
+
class BigIntVectorAccessor extends BaseArrowVectorAccessor<BigIntVector> {
public BigIntVectorAccessor(FieldVector vector) {
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorWriter.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorWriter.java
index e6a511a492..d30280077f 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorWriter.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/ArrowVectorWriter.java
@@ -50,6 +50,10 @@ public abstract class ArrowVectorWriter {
Map.entry(
IntegerType.class,
(fieldType, allocator, vector) -> new IntVectorWriter(fieldType,
allocator, vector)),
+ Map.entry(
+ TinyIntType.class,
+ (fieldType, allocator, vector) ->
+ new TinyIntVectorWriter(fieldType, allocator, vector)),
Map.entry(
BooleanType.class,
(fieldType, allocator, vector) ->
@@ -154,6 +158,7 @@ class FieldVectorCreator {
Map.ofEntries(
Map.entry(BooleanType.class, (dataType, timeZoneId) ->
ArrowType.Bool.INSTANCE),
Map.entry(IntegerType.class, (dataType, timeZoneId) -> new
ArrowType.Int(8 * 4, true)),
+ Map.entry(TinyIntType.class, (dataType, timeZoneId) -> new
ArrowType.Int(8 * 1, true)),
Map.entry(BigIntType.class, (dataType, timeZoneId) -> new
ArrowType.Int(8 * 8, true)),
Map.entry(
DoubleType.class,
@@ -284,6 +289,27 @@ class IntVectorWriter extends BaseVectorWriter<IntVector,
Integer> {
}
}
+class TinyIntVectorWriter extends BaseVectorWriter<TinyIntVector, Byte> {
+ public TinyIntVectorWriter(Type fieldType, BufferAllocator allocator,
FieldVector vector) {
+ super(vector);
+ }
+
+ @Override
+ protected Byte getValue(RowData rowData, int fieldIndex) {
+ return rowData.getByte(fieldIndex);
+ }
+
+ @Override
+ protected Byte getValue(ArrayData arrayData, int index) {
+ return arrayData.getByte(index);
+ }
+
+ @Override
+ protected void setValue(int index, Byte value) {
+ this.typedVector.setSafe(index, value);
+ }
+}
+
class BooleanVectorWriter extends BaseVectorWriter<BitVector, Boolean> {
public BooleanVectorWriter(Type fieldType, BufferAllocator allocator,
FieldVector vector) {
super(vector);
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertor.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertor.java
index 3abb06c511..04f6dc5769 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertor.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertor.java
@@ -25,9 +25,11 @@ import io.github.zhztheplayer.velox4j.type.Type;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.RowKind;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.TinyIntVector;
import org.apache.arrow.vector.table.Table;
import java.util.ArrayList;
@@ -35,11 +37,19 @@ import java.util.List;
/** Converter between velox RowVector and Flink RowData. */
public class FlinkRowToVLVectorConvertor {
+
+ // Matches the C++ kRowKindColumnName in
velox/experimental/stateful/RowKind.h.
+ // The merged RowVector carries per-row RowKind bytes as a trailing TINYINT
column
+ // with this name. Java appends it in fromRowData; C++ strips it via
+ // StreamRecord::create and re-appends it on output via
+ // StreamRecord::toMergedRowVector (only when the stream is not appendOnly).
+ public static final String ROW_KIND_COLUMN_NAME = "$row_kind";
+
public static RowVector fromRowData(
RowData row, BufferAllocator allocator, Session session, RowType
rowType) {
- List<FieldVector> arrowVectors = new ArrayList<>(rowType.size());
List<Type> fieldTypes = rowType.getChildren();
List<String> fieldNames = rowType.getNames();
+ List<FieldVector> arrowVectors = new ArrayList<>(rowType.size() + 1);
for (int i = 0; i < rowType.size(); i++) {
ArrowVectorWriter writer =
ArrowVectorWriter.create(fieldNames.get(i), fieldTypes.get(i),
allocator);
@@ -47,6 +57,11 @@ public class FlinkRowToVLVectorConvertor {
writer.finish();
arrowVectors.add(i, writer.getVector());
}
+ TinyIntVector rowKindVector = new TinyIntVector(ROW_KIND_COLUMN_NAME,
allocator);
+ rowKindVector.allocateNew(1);
+ rowKindVector.setSafe(0, row.getRowKind().toByteValue());
+ rowKindVector.setValueCount(1);
+ arrowVectors.add(rowKindVector);
return session.arrowOps().fromArrowTable(allocator, new
Table(arrowVectors));
}
@@ -62,14 +77,20 @@ public class FlinkRowToVLVectorConvertor {
// The result is StructVector
structVector = Arrow.toArrowVector(allocator, loadedVector);
final List<FieldVector> fieldVectors =
structVector.getChildrenFromFields();
- List<ArrowVectorAccessor> accessors =
buildArrowVectorAccessors(fieldVectors);
+ List<ArrowVectorAccessor> accessors =
+ buildArrowVectorAccessors(fieldVectors.subList(0, rowType.size()));
+ byte[] rowKinds = extractRowKindsIfPresent(fieldVectors, rowType.size(),
rowVector.getSize());
List<RowData> rowDatas = new ArrayList<>(rowVector.getSize());
for (int j = 0; j < rowVector.getSize(); j++) {
Object[] fieldValues = new Object[rowType.size()];
for (int i = 0; i < rowType.size(); i++) {
fieldValues[i] = accessors.get(i).get(j);
}
- rowDatas.add(GenericRowData.of(fieldValues));
+ if (rowKinds != null) {
+
rowDatas.add(GenericRowData.ofKind(RowKind.fromByteValue(rowKinds[j]),
fieldValues));
+ } else {
+ rowDatas.add(GenericRowData.of(fieldValues));
+ }
}
return rowDatas;
} finally {
@@ -85,6 +106,26 @@ public class FlinkRowToVLVectorConvertor {
}
}
+ private static byte[] extractRowKindsIfPresent(
+ List<FieldVector> fieldVectors, int schemaSize, int rowCount) {
+ if (fieldVectors.size() != schemaSize + 1) {
+ return null;
+ }
+ FieldVector lastVector = fieldVectors.get(schemaSize);
+ if (!ROW_KIND_COLUMN_NAME.equals(lastVector.getField().getName())) {
+ return null;
+ }
+ if (!(lastVector instanceof TinyIntVector)) {
+ return null;
+ }
+ TinyIntVector tinyInt = (TinyIntVector) lastVector;
+ byte[] rowKinds = new byte[rowCount];
+ for (int i = 0; i < rowCount; i++) {
+ rowKinds[i] = tinyInt.get(i);
+ }
+ return rowKinds;
+ }
+
private static List<ArrowVectorAccessor>
buildArrowVectorAccessors(List<FieldVector> vectors) {
List<ArrowVectorAccessor> accessors = new ArrayList<>(vectors.size());
for (int i = 0; i < vectors.size(); ++i) {
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertorTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertorTest.java
new file mode 100644
index 0000000000..36558d94e5
--- /dev/null
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertorTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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.gluten.vectorized;
+
+import org.apache.gluten.table.runtime.stream.common.Velox4jEnvironment;
+
+import io.github.zhztheplayer.velox4j.Velox4j;
+import io.github.zhztheplayer.velox4j.data.RowVector;
+import io.github.zhztheplayer.velox4j.memory.AllocationListener;
+import io.github.zhztheplayer.velox4j.memory.MemoryManager;
+import io.github.zhztheplayer.velox4j.session.Session;
+
+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.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.types.RowKind;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.TinyIntVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.table.Table;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class FlinkRowToVLVectorConvertorTest {
+ private static MemoryManager memoryManager;
+ private static Session session;
+ private static BufferAllocator allocator;
+ private static RowType flinkRowType;
+ private static io.github.zhztheplayer.velox4j.type.RowType veloxRowType;
+
+ @BeforeAll
+ public static void initializeVelox() {
+ Velox4jEnvironment.initializeOnce();
+ memoryManager = MemoryManager.create(AllocationListener.NOOP);
+ session = Velox4j.newSession(memoryManager);
+ allocator = new RootAllocator(Long.MAX_VALUE);
+ flinkRowType =
+ RowType.of(
+ new LogicalType[] {new IntType(), new
VarCharType(VarCharType.MAX_LENGTH)},
+ new String[] {"id", "name"});
+ veloxRowType =
+ (io.github.zhztheplayer.velox4j.type.RowType)
+ org.apache.gluten.util.LogicalTypeConverter.toVLType(flinkRowType);
+ }
+
+ @AfterAll
+ public static void tearDownVelox() {
+ if (allocator != null) {
+ allocator.close();
+ allocator = null;
+ }
+ if (session != null) {
+ session.close();
+ session = null;
+ }
+ if (memoryManager != null) {
+ memoryManager.close();
+ memoryManager = null;
+ }
+ }
+
+ @Test
+ public void testToRowDataReadsRowKindFromMergedRowVector() {
+ RowKind[] kinds = {
+ RowKind.INSERT, RowKind.UPDATE_BEFORE, RowKind.UPDATE_AFTER,
RowKind.DELETE,
+ };
+ List<FieldVector> arrowVectors = new ArrayList<>(3);
+ IntVector idVec = new IntVector("id", allocator);
+ idVec.allocateNew(4);
+ VarCharVector nameVec = new VarCharVector("name", allocator);
+ nameVec.allocateNew(4);
+ TinyIntVector rowKindVec =
+ new TinyIntVector(
+ FlinkRowToVLVectorConvertor.ROW_KIND_COLUMN_NAME,
+ new FieldType(false, new ArrowType.Int(8, true), null),
+ allocator);
+ rowKindVec.allocateNew(4);
+ for (int i = 0; i < kinds.length; i++) {
+ idVec.setSafe(i, 100 + i);
+ nameVec.setSafe(i, ("u" +
i).getBytes(java.nio.charset.StandardCharsets.UTF_8));
+ rowKindVec.setSafe(i, kinds[i].toByteValue());
+ }
+ idVec.setValueCount(4);
+ nameVec.setValueCount(4);
+ rowKindVec.setValueCount(4);
+ arrowVectors.add(idVec);
+ arrowVectors.add(nameVec);
+ arrowVectors.add(rowKindVec);
+ RowVector rv = session.arrowOps().fromArrowTable(allocator, new
Table(arrowVectors));
+ try {
+ List<RowData> rows = FlinkRowToVLVectorConvertor.toRowData(rv,
allocator, veloxRowType);
+ assertThat(rows).hasSize(kinds.length);
+ for (int i = 0; i < kinds.length; i++) {
+ assertThat(rows.get(i).getRowKind()).isEqualTo(kinds[i]);
+ assertThat(rows.get(i).getInt(0)).isEqualTo(100 + i);
+
assertThat(rows.get(i).getString(1)).isEqualTo(StringData.fromString("u" + i));
+ }
+ } finally {
+ rv.close();
+ idVec.close();
+ nameVec.close();
+ rowKindVec.close();
+ }
+ }
+
+ @Test
+ public void testToRowDataFallsBackToInsertWhenNoRowKindColumn() {
+ RowData input = GenericRowData.of(42, StringData.fromString("alice"));
+ RowVector rv = FlinkRowToVLVectorConvertor.fromRowData(input, allocator,
session, veloxRowType);
+ try {
+ List<RowData> rows = FlinkRowToVLVectorConvertor.toRowData(rv,
allocator, veloxRowType);
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getRowKind()).isEqualTo(RowKind.INSERT);
+ assertThat(rows.get(0).getInt(0)).isEqualTo(42);
+
assertThat(rows.get(0).getString(1)).isEqualTo(StringData.fromString("alice"));
+ } finally {
+ rv.close();
+ }
+ }
+
+ @Test
+ public void testFromRowDataAppendsRowKindColumnForAllKinds() {
+ RowKind[] kinds = {
+ RowKind.INSERT, RowKind.UPDATE_BEFORE, RowKind.UPDATE_AFTER,
RowKind.DELETE,
+ };
+ for (RowKind kind : kinds) {
+ RowData input =
+ GenericRowData.ofKind(kind, 7, StringData.fromString("row-" +
kind.shortString()));
+ RowVector rv =
+ FlinkRowToVLVectorConvertor.fromRowData(input, allocator, session,
veloxRowType);
+ try {
+ List<RowData> rows = FlinkRowToVLVectorConvertor.toRowData(rv,
allocator, veloxRowType);
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getRowKind()).isEqualTo(kind);
+ assertThat(rows.get(0).getInt(0)).isEqualTo(7);
+ assertThat(rows.get(0).getString(1))
+ .isEqualTo(StringData.fromString("row-" + kind.shortString()));
+ } finally {
+ rv.close();
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
