This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new d851e0a897a [FLINK-41040][state-processor-api] Fix binary row
conversion.
d851e0a897a is described below
commit d851e0a897a68655f061b096bb70c6d245b10adc
Author: Muhammad Junaid Muzammil <[email protected]>
AuthorDate: Tue Oct 20 04:36:34 2026 -0700
[FLINK-41040][state-processor-api] Fix binary row conversion.
---
.../flink/state/table/StateValueConverter.java | 32 ++++++
.../flink/state/table/StateValueConverterTest.java | 113 +++++++++++++++++++++
2 files changed, 145 insertions(+)
diff --git
a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/StateValueConverter.java
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/StateValueConverter.java
index 40b9c0d8b7e..19781fe35bb 100644
---
a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/StateValueConverter.java
+++
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/StateValueConverter.java
@@ -71,6 +71,7 @@ class StateValueConverter implements java.io.Serializable {
// field/getter is only ever resolved via reflection once per class
instead of on every row.
private transient Map<Tuple2<Class, String>, Optional<Field>>
classFieldCache;
private transient Map<Tuple2<Class, String>, Optional<Method>>
classMethodCache;
+ private transient Map<RowType, RowData.FieldGetter[]> rowFieldGetterCache;
private Map<Tuple2<Class, String>, Optional<Field>> classFieldCache() {
if (classFieldCache == null) {
@@ -86,6 +87,22 @@ class StateValueConverter implements java.io.Serializable {
return classMethodCache;
}
+ private RowData.FieldGetter[] rowFieldGetters(RowType rowType) {
+ if (rowFieldGetterCache == null) {
+ rowFieldGetterCache = new ConcurrentHashMap<>();
+ }
+ return rowFieldGetterCache.computeIfAbsent(
+ rowType,
+ type -> {
+ final RowData.FieldGetter[] getters =
+ new RowData.FieldGetter[type.getFieldCount()];
+ for (int i = 0; i < getters.length; i++) {
+ getters[i] =
RowData.createFieldGetter(type.getTypeAt(i), i);
+ }
+ return getters;
+ });
+ }
+
/**
* Reads the value from a VALUE-shaped state, dispatching to {@code
ValueState.value()}, {@code
* ReducingState.get()} or {@code AggregatingState.get()} depending on
{@code actualStateKind}
@@ -193,6 +210,21 @@ class StateValueConverter implements java.io.Serializable {
}
return result;
}
+ if (object instanceof RowData) {
+ final RowData sourceRow = (RowData) object;
+ final RowType rowType = (RowType) logicalType;
+ final RowData.FieldGetter[] getters =
rowFieldGetters(rowType);
+ final GenericRowData result =
+ new GenericRowData(RowKind.INSERT,
rowType.getFieldCount());
+ for (int i = 0; i < rowType.getFieldCount(); i++) {
+ result.setField(
+ i,
+ i < sourceRow.getArity()
+ ? getters[i].getFieldOrNull(sourceRow)
+ : null);
+ }
+ return result;
+ }
return convertToRow(object, logicalType);
default:
return InternalTypeConverter.toInternal(object, logicalType);
diff --git
a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/StateValueConverterTest.java
b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/StateValueConverterTest.java
new file mode 100644
index 00000000000..646923b2869
--- /dev/null
+++
b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/StateValueConverterTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.flink.state.table;
+
+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.data.binary.BinaryRowData;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.types.logical.BooleanType;
+import org.apache.flink.table.types.logical.IntType;
+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.apache.flink.types.RowKind;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class StateValueConverterTest {
+
+ private final StateValueConverter converter = new StateValueConverter();
+
+ @Test
+ void convertsBinaryStateKey() {
+ final RowType type = RowType.of(new IntType(false));
+ final BinaryRowData source = new
RowDataSerializer(type).toBinaryRow(GenericRowData.of(42));
+
+ final RowData result = (RowData) converter.getValue(type, source);
+
+ assertThat(result.getInt(0)).isEqualTo(42);
+ assertThat(result.getRowKind()).isEqualTo(RowKind.INSERT);
+ }
+
+ @Test
+ void convertsBinaryMapKey() {
+ final RowType type =
+ RowType.of(new IntType(false), new BooleanType(), new
TimestampType(3));
+ final TimestampData timestamp =
TimestampData.fromEpochMillis(123456789L);
+ final BinaryRowData source =
+ new RowDataSerializer(type).toBinaryRow(GenericRowData.of(7,
true, timestamp));
+
+ final RowData result = (RowData) converter.getValue(type, source);
+
+ assertThat(result.getInt(0)).isEqualTo(7);
+ assertThat(result.getBoolean(1)).isTrue();
+ assertThat(result.getTimestamp(2, 3)).isEqualTo(timestamp);
+ }
+
+ @Test
+ void preservesNullBinaryFields() {
+ final RowType type = RowType.of(new IntType(), new VarCharType(), new
TimestampType(3));
+ final BinaryRowData source =
+ new
RowDataSerializer(type).toBinaryRow(GenericRowData.of(null, null, null));
+
+ final RowData result = (RowData) converter.getValue(type, source);
+
+ assertThat(result.isNullAt(0)).isTrue();
+ assertThat(result.isNullAt(1)).isTrue();
+ assertThat(result.isNullAt(2)).isTrue();
+ }
+
+ @Test
+ void readsNestedBinaryRows() {
+ final RowType nestedType = RowType.of(new IntType(), new
VarCharType());
+ final RowType type = RowType.of(new IntType(), nestedType);
+ final BinaryRowData source =
+ new RowDataSerializer(type)
+ .toBinaryRow(
+ GenericRowData.of(
+ 1, GenericRowData.of(2,
StringData.fromString("nested"))));
+
+ final RowData result = (RowData) converter.getValue(type, source);
+ final RowData nested = result.getRow(1, 2);
+
+ assertThat(result.getInt(0)).isEqualTo(1);
+ assertThat(nested.getInt(0)).isEqualTo(2);
+ assertThat(nested.getString(1).toString()).isEqualTo("nested");
+ }
+
+ @Test
+ void preservesGenericRowNormalizationAndMissingFields() {
+ final RowType type = RowType.of(new IntType(), new VarCharType(), new
BooleanType());
+ final GenericRowData source = GenericRowData.of(3,
StringData.fromString("existing"));
+ source.setRowKind(RowKind.UPDATE_AFTER);
+
+ final RowData result = (RowData) converter.getValue(type, source);
+
+ assertThat(result).isNotSameAs(source);
+ assertThat(result.getRowKind()).isEqualTo(RowKind.INSERT);
+ assertThat(result.getInt(0)).isEqualTo(3);
+ assertThat(result.getString(1).toString()).isEqualTo("existing");
+ assertThat(result.isNullAt(2)).isTrue();
+ }
+}
