Skip to content

Flink: Move write from AppenderFactory to FileWriterFactory #14271

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pvary merged 2 commits into from
Oct 14, 2025
Merged

Flink: Move write from AppenderFactory to FileWriterFactory #14271

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
209a9a5
Flink: Move write from Appenders to Writers
pvary Oct 7, 2025
6f79556
Review comments
pvary Oct 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 37 additions & 3 deletions core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ public abstract class BaseTaskWriter<T> implements TaskWriter<T> {
private final PartitionSpec spec;
private final FileFormat format;
private final FileAppenderFactory<T> appenderFactory;
private final FileWriterFactory<T> writerFactory;
private final OutputFileFactory fileFactory;
private final FileIO io;
private final long targetFileSize;
Expand All @@ -71,6 +72,23 @@ protected BaseTaskWriter(
this.spec = spec;
this.format = format;
this.appenderFactory = appenderFactory;
this.writerFactory = null;
this.fileFactory = fileFactory;
this.io = io;
this.targetFileSize = targetFileSize;
}

protected BaseTaskWriter(

@stevenzwu stevenzwu Oct 7, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess this is an approach with minimal effort to move Flink to FileWriterFactory.

Otherwise, the intention is probably for Flink and Kafka to move away from BaseTaskWriter as well. E.g., Spark is using FanoutDataWriter and ClusteredDataWriter.

Should we move Flink to use PartitioningWriter? might be a bigger change though.

@pvary pvary Oct 8, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What we are missing is the BaseEqualityDeltaWriter - RowDataDeltaWriter which keeps track of the previously written rows and create a position delete if it is deleted again.

This could be done, but I would postpone it to another PR.

This PR would allow us to test the FileFormat API with Flink and also not expose the WriteBuilder on the FormatModelRegistry

@stevenzwu stevenzwu Oct 9, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am fine with this approach. It doesn't require a big effort to migrate Flink to FileWriterFactory as an intermediate step, although if-else isn't perfect.

@pvary pvary Oct 10, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we deprecate the AppenderFactory then we can remove the if statements.

PartitionSpec spec,
FileFormat format,
FileWriterFactory<T> writerFactory,
OutputFileFactory fileFactory,
FileIO io,
long targetFileSize) {
this.spec = spec;
this.format = format;
this.appenderFactory = null;
this.writerFactory = writerFactory;
this.fileFactory = fileFactory;
this.io = io;
this.targetFileSize = targetFileSize;
Expand Down Expand Up @@ -138,7 +156,15 @@ protected BaseEqualityDeltaWriter(
this.eqDeleteWriter = new RollingEqDeleteWriter(partition);
this.posDeleteWriter =
new SortingPositionOnlyDeleteWriter<>(
() -> appenderFactory.newPosDeleteWriter(newOutputFile(partition), format, partition),
() -> {
if (writerFactory != null) {
return writerFactory.newPositionDeleteWriter(
newOutputFile(partition), spec, partition);
} else {
return appenderFactory.newPosDeleteWriter(
newOutputFile(partition), format, partition);
}
},
deleteGranularity);
this.insertedRowMap = StructLikeMap.create(deleteSchema.asStruct());
}
Expand Down Expand Up @@ -388,7 +414,11 @@ public RollingFileWriter(StructLike partitionKey) {

@Override
DataWriter<T> newWriter(EncryptedOutputFile file, StructLike partitionKey) {
return appenderFactory.newDataWriter(file, format, partitionKey);
if (writerFactory != null) {
return writerFactory.newDataWriter(file, spec, partitionKey);
} else {
return appenderFactory.newDataWriter(file, format, partitionKey);
}
}

@Override
Expand All @@ -414,7 +444,11 @@ protected class RollingEqDeleteWriter extends BaseRollingWriter<EqualityDeleteWr

@Override
EqualityDeleteWriter<T> newWriter(EncryptedOutputFile file, StructLike partitionKey) {
return appenderFactory.newEqDeleteWriter(file, format, partitionKey);
if (writerFactory != null) {
return writerFactory.newEqualityDeleteWriter(file, spec, partitionKey);
} else {
return appenderFactory.newEqDeleteWriter(file, format, partitionKey);
}
}

@Override
Expand Down
10 changes: 10 additions & 0 deletions core/src/main/java/org/apache/iceberg/io/PartitionedFanoutWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,16 @@ protected PartitionedFanoutWriter(
super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
}

protected PartitionedFanoutWriter(
PartitionSpec spec,
FileFormat format,
FileWriterFactory<T> fileWriterFactory,
OutputFileFactory fileFactory,
FileIO io,
long targetFileSize) {
super(spec, format, fileWriterFactory, fileFactory, io, targetFileSize);
}

/**
* Create a PartitionKey from the values in row.
*
Expand Down
10 changes: 10 additions & 0 deletions core/src/main/java/org/apache/iceberg/io/PartitionedWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,16 @@ protected PartitionedWriter(
super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
}

protected PartitionedWriter(

@stevenzwu stevenzwu Oct 7, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

looks like PartitionedWriter is not used in Iceberg repo anymore

@pvary pvary Oct 8, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, it is not changed with this PR.
But, yeah, we might want to deprecate it.

PartitionSpec spec,
FileFormat format,
FileWriterFactory<T> fileWriterFactory,
OutputFileFactory fileFactory,
FileIO io,
long targetFileSize) {
super(spec, format, fileWriterFactory, fileFactory, io, targetFileSize);
}

/**
* Create a PartitionKey from the values in row.
*
Expand Down
11 changes: 11 additions & 0 deletions core/src/main/java/org/apache/iceberg/io/UnpartitionedWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,17 @@ public UnpartitionedWriter(
currentWriter = new RollingFileWriter(null);
}

public UnpartitionedWriter(
PartitionSpec spec,
FileFormat format,
FileWriterFactory<T> fileWriterFactory,
OutputFileFactory fileFactory,
FileIO io,
long targetFileSize) {
super(spec, format, fileWriterFactory, fileFactory, io, targetFileSize);
currentWriter = new RollingFileWriter(null);
}

@Override
public void write(T record) throws IOException {
currentWriter.write(record);
Expand Down
39 changes: 38 additions & 1 deletion data/src/main/java/org/apache/iceberg/data/BaseFileWriterFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.apache.iceberg.data;

import java.io.IOException;
import java.io.Serializable;
import java.io.UncheckedIOException;
import java.util.Map;
import org.apache.iceberg.FileFormat;
Expand All @@ -37,9 +38,10 @@
import org.apache.iceberg.io.FileWriterFactory;
import org.apache.iceberg.orc.ORC;
import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;

/** A base writer factory to be extended by query engine integrations. */
public abstract class BaseFileWriterFactory<T> implements FileWriterFactory<T> {
public abstract class BaseFileWriterFactory<T> implements FileWriterFactory<T>, Serializable {

@pvary pvary Oct 7, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

needed for Flink serialization

@stevenzwu stevenzwu Oct 8, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Checked the child classes. SparkFileWriterFactory is probably ok, as the Table object was extracted from a broadcast variable for SerializableTable. But GenericFileWriterFactory ma be constructed with a regular and may not be serializable.

Since FlinkFileWriterFactory is already marked as serializable. maybe we don't need this change?

@pvary pvary Oct 9, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since FlinkFileWriterFactory is already marked as serializable. maybe we don't need this change?

I had test failures. Flink was not able to serialize/deserialize the factory, because the base class was not serializable.

GenericFileWriterFactory ma be constructed with a regular and may not be serializable.

BaseTable implements writeReplace, which makes sure that the table object is always serializable

  Object writeReplace() {
    return SerializableTable.copyOf(this);
  }

Added a test anyways.

@stevenzwu stevenzwu Oct 9, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ah. I didn't notice that BaseTable implements writeReplace. thanks for explaining

private final Table table;
private final FileFormat dataFileFormat;
private final Schema dataSchema;
Expand All @@ -49,7 +51,32 @@ public abstract class BaseFileWriterFactory<T> implements FileWriterFactory<T> {
private final Schema equalityDeleteRowSchema;
private final SortOrder equalityDeleteSortOrder;
private final Schema positionDeleteRowSchema;
private final Map<String, String> writerProperties;

protected BaseFileWriterFactory(
Table table,
FileFormat dataFileFormat,
Schema dataSchema,
SortOrder dataSortOrder,
FileFormat deleteFileFormat,
int[] equalityFieldIds,
Schema equalityDeleteRowSchema,
SortOrder equalityDeleteSortOrder,
Schema positionDeleteRowSchema,
Map<String, String> writerProperties) {
this.table = table;
this.dataFileFormat = dataFileFormat;
this.dataSchema = dataSchema;
this.dataSortOrder = dataSortOrder;
this.deleteFileFormat = deleteFileFormat;
this.equalityFieldIds = equalityFieldIds;
this.equalityDeleteRowSchema = equalityDeleteRowSchema;
this.equalityDeleteSortOrder = equalityDeleteSortOrder;
this.positionDeleteRowSchema = positionDeleteRowSchema;
this.writerProperties = writerProperties;
}

@Deprecated
protected BaseFileWriterFactory(
Table table,
FileFormat dataFileFormat,
Expand All @@ -69,6 +96,7 @@ protected BaseFileWriterFactory(
this.equalityDeleteRowSchema = equalityDeleteRowSchema;
this.equalityDeleteSortOrder = equalityDeleteSortOrder;
this.positionDeleteRowSchema = positionDeleteRowSchema;
this.writerProperties = ImmutableMap.of();
}

protected abstract void configureDataWrite(Avro.DataWriteBuilder builder);
Expand Down Expand Up @@ -103,6 +131,7 @@ public DataWriter<T> newDataWriter(
Avro.writeData(file)
.schema(dataSchema)
.setAll(properties)
.setAll(writerProperties)
.metricsConfig(metricsConfig)
.withSpec(spec)
.withPartition(partition)
Expand All @@ -119,6 +148,7 @@ public DataWriter<T> newDataWriter(
Parquet.writeData(file)
.schema(dataSchema)
.setAll(properties)
.setAll(writerProperties)
.metricsConfig(metricsConfig)
.withSpec(spec)
.withPartition(partition)
Expand All @@ -135,6 +165,7 @@ public DataWriter<T> newDataWriter(
ORC.writeData(file)
.schema(dataSchema)
.setAll(properties)
.setAll(writerProperties)
.metricsConfig(metricsConfig)
.withSpec(spec)
.withPartition(partition)
Expand Down Expand Up @@ -168,6 +199,7 @@ public EqualityDeleteWriter<T> newEqualityDeleteWriter(
Avro.DeleteWriteBuilder avroBuilder =
Avro.writeDeletes(file)
.setAll(properties)
.setAll(writerProperties)
.metricsConfig(metricsConfig)
.rowSchema(equalityDeleteRowSchema)
.equalityFieldIds(equalityFieldIds)
Expand All @@ -185,6 +217,7 @@ public EqualityDeleteWriter<T> newEqualityDeleteWriter(
Parquet.DeleteWriteBuilder parquetBuilder =
Parquet.writeDeletes(file)
.setAll(properties)
.setAll(writerProperties)
.metricsConfig(metricsConfig)
.rowSchema(equalityDeleteRowSchema)
.equalityFieldIds(equalityFieldIds)
Expand All @@ -202,6 +235,7 @@ public EqualityDeleteWriter<T> newEqualityDeleteWriter(
ORC.DeleteWriteBuilder orcBuilder =
ORC.writeDeletes(file)
.setAll(properties)
.setAll(writerProperties)
.metricsConfig(metricsConfig)
.rowSchema(equalityDeleteRowSchema)
.equalityFieldIds(equalityFieldIds)
Expand Down Expand Up @@ -237,6 +271,7 @@ public PositionDeleteWriter<T> newPositionDeleteWriter(
Avro.DeleteWriteBuilder avroBuilder =
Avro.writeDeletes(file)
.setAll(properties)
.setAll(writerProperties)
.metricsConfig(metricsConfig)
.rowSchema(positionDeleteRowSchema)
.withSpec(spec)
Expand All @@ -252,6 +287,7 @@ public PositionDeleteWriter<T> newPositionDeleteWriter(
Parquet.DeleteWriteBuilder parquetBuilder =
Parquet.writeDeletes(file)
.setAll(properties)
.setAll(writerProperties)
.metricsConfig(metricsConfig)
.rowSchema(positionDeleteRowSchema)
.withSpec(spec)
Expand All @@ -267,6 +303,7 @@ public PositionDeleteWriter<T> newPositionDeleteWriter(
ORC.DeleteWriteBuilder orcBuilder =
ORC.writeDeletes(file)
.setAll(properties)
.setAll(writerProperties)
.metricsConfig(metricsConfig)
.rowSchema(positionDeleteRowSchema)
.withSpec(spec)
Expand Down
12 changes: 12 additions & 0 deletions data/src/test/java/org/apache/iceberg/io/TestFileWriterFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import static org.apache.iceberg.MetadataColumns.DELETE_FILE_POS;
import static org.apache.iceberg.MetadataColumns.DELETE_FILE_ROW_FIELD_NAME;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNoException;
import static org.assertj.core.api.Assumptions.assumeThat;

import java.io.File;
Expand Down Expand Up @@ -55,8 +56,10 @@
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.CharSequenceSet;
import org.apache.iceberg.util.Pair;
import org.apache.iceberg.util.SerializationUtil;
import org.apache.iceberg.util.StructLikeSet;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;

Expand Down Expand Up @@ -406,6 +409,15 @@ public void testPositionDeleteWriterMultipleDataFiles() throws IOException {
assertThat(actualRowSet("*")).isEqualTo(toSet(expectedRows));
}

@Test
void testSerialization() {
FileWriterFactory<T> writerFactory = newWriterFactory(table.schema());
assertThatNoException().isThrownBy(() -> SerializationUtil.serializeToBytes(writerFactory));

byte[] serialized = SerializationUtil.serializeToBytes(writerFactory);
assertThatNoException().isThrownBy(() -> SerializationUtil.deserializeFromBytes(serialized));
}

private DataFile writeData(
FileWriterFactory<T> writerFactory, List<T> rows, PartitionSpec spec, StructLike partitionKey)
throws IOException {
Expand Down
6 changes: 3 additions & 3 deletions flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,8 +32,8 @@
import org.apache.iceberg.flink.RowDataWrapper;
import org.apache.iceberg.flink.data.RowDataProjection;
import org.apache.iceberg.io.BaseTaskWriter;
import org.apache.iceberg.io.FileAppenderFactory;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.FileWriterFactory;
import org.apache.iceberg.io.OutputFileFactory;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.types.TypeUtil;
Expand All @@ -50,15 +50,15 @@ abstract class BaseDeltaTaskWriter extends BaseTaskWriter<RowData> {
BaseDeltaTaskWriter(
PartitionSpec spec,
FileFormat format,
FileAppenderFactory<RowData> appenderFactory,
FileWriterFactory<RowData> fileWriterFactory,
OutputFileFactory fileFactory,
FileIO io,
long targetFileSize,
Schema schema,
RowType flinkSchema,
Set<Integer> equalityFieldIds,
boolean upsert) {
super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
super(spec, format, fileWriterFactory, fileFactory, io, targetFileSize);
this.schema = schema;
this.deleteSchema = TypeUtil.select(schema, Sets.newHashSet(equalityFieldIds));
this.wrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
Expand Down
5 changes: 5 additions & 0 deletions flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,11 @@
import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;

/**
* @deprecated Deprecated as of 1.11.0 in favor of {@link FlinkFileWriterFactory}. This class will
* be removed in the 1.12.0.
*/
@Deprecated
public class FlinkAppenderFactory implements FileAppenderFactory<RowData>, Serializable {
private final Schema schema;
private final RowType flinkSchema;
Expand Down
Loading