You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by pv...@apache.org on 2026/10/18 18:48:55 UTC
(iceberg) branch main updated: Flink: Backport deep copy RowData in DynamicRecordInternalSerializer#copy to Flink 2.2 (#18547)
This is an automated email from the ASF dual-hosted git repository.
pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 318e01de78 Flink: Backport deep copy RowData in DynamicRecordInternalSerializer#copy to Flink 2.2 (#18547)
318e01de78 is described below
commit 318e01de7830e8d70df21fd17a5b5ab7f36f89d8
Author: openinx <op...@gmail.com>
AuthorDate: Sun Oct 18 11:48:49 2026 -0700
Flink: Backport deep copy RowData in DynamicRecordInternalSerializer#copy to Flink 2.2 (#18547)
backports #18497
---
.../dynamic/DynamicRecordInternalSerializer.java | 8 ++-
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 45 ++++++++++++++
.../TestDynamicRecordInternalSerializer.java | 68 ++++++++++++++++++++++
3 files changed, 119 insertions(+), 2 deletions(-)
diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java
index 0d758ace1b..e9e7daef46 100644
--- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java
+++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java
@@ -215,11 +215,13 @@ class DynamicRecordInternalSerializer extends TypeSerializer<DynamicRecordIntern
@Override
public DynamicRecordInternal copy(DynamicRecordInternal from) {
+ RowDataSerializer rowDataSerializer =
+ serializerCache.serializer(from.tableName(), from.schema(), from.spec());
return new DynamicRecordInternal(
from.tableName(),
from.branch(),
from.schema(),
- from.rowData(),
+ rowDataSerializer.copy(from.rowData()),
from.spec(),
from.writerKey(),
from.upsertMode(),
@@ -228,12 +230,14 @@ class DynamicRecordInternalSerializer extends TypeSerializer<DynamicRecordIntern
@Override
public DynamicRecordInternal copy(DynamicRecordInternal from, DynamicRecordInternal reuse) {
+ RowDataSerializer rowDataSerializer =
+ serializerCache.serializer(from.tableName(), from.schema(), from.spec());
reuse.setTableName(from.tableName());
reuse.setBranch(from.branch());
reuse.setSchema(from.schema());
reuse.setSpec(from.spec());
reuse.setWriterKey(from.writerKey());
- reuse.setRowData(from.rowData());
+ reuse.setRowData(rowDataSerializer.copy(from.rowData()));
reuse.setUpsertMode(from.upsertMode());
reuse.setEqualityFieldIds(from.equalityFields());
return reuse;
diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index 2abcf9ce01..34a74cdf1b 100644
--- a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -24,6 +24,7 @@ import static org.assertj.core.api.Assertions.fail;
import java.io.IOException;
import java.io.Serializable;
+import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
@@ -407,6 +408,50 @@ class TestDynamicIcebergSink extends TestFlinkIcebergSinkBase {
assertThat(generatorSSGApplied).isTrue();
}
+ @Test
+ void forwardOrcWritePreservesReusedBinaryBuffer() throws Exception {
+ Schema schema = new Schema(Types.NestedField.required(1, "data", Types.BinaryType.get()));
+ TableIdentifier identifier = TableIdentifier.of(DATABASE, "t1");
+ Table table = CATALOG_EXTENSION.catalog().createTable(identifier, schema);
+ table
+ .updateProperties()
+ .set(TableProperties.DEFAULT_FILE_FORMAT, FileFormat.ORC.name())
+ .commit();
+ env.setParallelism(1);
+ env.getConfig().disableObjectReuse();
+ List<Byte> values = List.of((byte) 1, (byte) 2);
+
+ DynamicIcebergSink.forInput(env.fromData(0))
+ .generator(
+ (ignored, out) -> {
+ byte[] buffer = new byte[1];
+ GenericRowData row = GenericRowData.of(buffer);
+ for (byte value : values) {
+ buffer[0] = value;
+ out.collect(
+ new DynamicRecord(
+ TableIdentifier.of(DATABASE, "t1"),
+ SnapshotRef.MAIN_BRANCH,
+ schema,
+ row,
+ PartitionSpec.unpartitioned()));
+ }
+ })
+ .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+ .writeParallelism(1)
+ .immediateTableUpdate(true)
+ .append();
+
+ env.execute();
+
+ table.refresh();
+ try (CloseableIterable<Record> records = IcebergGenerics.read(table).build()) {
+ assertThat(records)
+ .extracting(record -> record.get(0, ByteBuffer.class).get())
+ .containsExactlyInAnyOrderElementsOf(values);
+ }
+ }
+
@Test
void testForwardWrite() throws Exception {
runForwardWriteTest(new ForwardGenerator());
diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordInternalSerializer.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordInternalSerializer.java
index 388ff30333..5667583599 100644
--- a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordInternalSerializer.java
+++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordInternalSerializer.java
@@ -18,16 +18,26 @@
*/
package org.apache.iceberg.flink.sink.dynamic;
+import static org.apache.iceberg.types.Types.NestedField.required;
import static org.assertj.core.api.Assertions.assertThat;
import java.io.IOException;
+import java.util.Collections;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.RowKind;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
import org.apache.iceberg.flink.HadoopCatalogExtension;
+import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
public class TestDynamicRecordInternalSerializer {
@@ -97,6 +107,64 @@ public class TestDynamicRecordInternalSerializer {
.isTrue();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void copyPreservesValues(boolean reuse) {
+ Schema schema = new Schema(required(1, "id", Types.LongType.get()));
+ long value = 123L;
+ GenericRowData row = GenericRowData.of(value);
+ DynamicRecordInternal copied = copy(schema, row, reuse);
+
+ row.setField(0, value + 1);
+
+ assertThat(copied.rowData().getLong(0)).isEqualTo(value);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void copyPreservesNestedValues(boolean reuse) {
+ Schema schema =
+ new Schema(
+ required(
+ 1, "nested", Types.StructType.of(required(2, "data", Types.BinaryType.get()))));
+ byte[] value = new byte[] {1, 2, 3};
+ byte[] expected = value.clone();
+ GenericRowData row = GenericRowData.of(GenericRowData.of(value));
+ DynamicRecordInternal copied = copy(schema, row, reuse);
+
+ value[0] += 1;
+
+ assertThat(copied.rowData().getRow(0, 1).getBinary(0)).isEqualTo(expected);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void copyPreservesRowKind(boolean reuse) {
+ Schema schema = new Schema(required(1, "id", Types.LongType.get()));
+ GenericRowData row = GenericRowData.of(123L);
+ row.setRowKind(RowKind.UPDATE_BEFORE);
+ DynamicRecordInternal copied = copy(schema, row, reuse);
+
+ row.setRowKind(RowKind.UPDATE_AFTER);
+
+ assertThat(copied.rowData().getRowKind()).isEqualTo(RowKind.UPDATE_BEFORE);
+ }
+
+ private DynamicRecordInternal copy(Schema schema, RowData row, boolean reuse) {
+ DynamicRecordInternalSerializer serializer = createSerializer();
+ DynamicRecordInternal record =
+ new DynamicRecordInternal(
+ "table",
+ "main",
+ schema,
+ row,
+ PartitionSpec.unpartitioned(),
+ 0,
+ false,
+ Collections.emptySet());
+ return reuse ? serializer.copy(record, serializer.createInstance()) : serializer.copy(record);
+ }
+
private DynamicRecordInternalSerializer createSerializer() {
return new DynamicRecordInternalSerializer(
new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1), true);