Flink: Move write from AppenderFactory to FileWriterFactory - #14271
Conversation
|
|
||
| /** 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 { |
There was a problem hiding this comment.
needed for Flink serialization
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
ah. I didn't notice that BaseTable implements writeReplace. thanks for explaining
|
CC: @stevenzwu, @Guosmilesmile, @mxm |
| super(spec, format, appenderFactory, fileFactory, io, targetFileSize); | ||
| } | ||
|
|
||
| protected PartitionedWriter( |
There was a problem hiding this comment.
looks like PartitionedWriter is not used in Iceberg repo anymore
There was a problem hiding this comment.
Yeah, it is not changed with this PR.
But, yeah, we might want to deprecate it.
| this.targetFileSize = targetFileSize; | ||
| } | ||
|
|
||
| protected BaseTaskWriter( |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
If we deprecate the AppenderFactory then we can remove the if statements.
stevenzwu
left a comment
There was a problem hiding this comment.
LGTM. just a nit comment on annotations for the public class
| import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; | ||
|
|
||
| class FlinkFileWriterFactory extends BaseFileWriterFactory<RowData> implements Serializable { | ||
| public class FlinkFileWriterFactory extends BaseFileWriterFactory<RowData> implements Serializable { |
There was a problem hiding this comment.
nit: looks like the public is only needed for SimpleDataUtil and other test class. Should we add @Internal and @VisibleForTest annotations?
There was a problem hiding this comment.
As discussed offline, this needs to be public so we can deprecate the FlinkFileAppenderFactory, and users of that could migrate using the new class.
|
Merged to main. |
@aokolnychyi created the
FileWriterFactoryinterfaces to write Data/PositionDelete/EqualityDelete Iceberg files.Spark moved to the new API, but Flink never started to use it.
During the FileFormat API discussions some raised the question if we want to take the move now, before migrating to the new API. This PR attempts to do this.
There are 2 main parts:
BaseTaskWriterto use bothFileAppenderFactoryandFileWriterFactorywhichever is availablewriterPropertiesto theBaseFileWriterFactory, to allow migrating the Flink functionality where the users of theFlinkFileAppenderFactorycan provide additional configurations to the appender.BaseDeltaTaskWriterto use theFlinkFileWriterFactory.FlinkFileWriterFactory.2 usages for the
FlinkAppenderFactoryremains:TestFlinkAppenderFactory- obviously we will remove when the deprecation period is finishedTestFlinkMergingMetrics- we should move this FileFormat TCK