You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "pvary (via GitHub)" <gi...@apache.org> on 2025/10/07 13:52:25 UTC
[PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
pvary opened a new pull request, #14271:
URL: https://github.com/apache/iceberg/pull/14271
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2416110435
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java:
##########
@@ -240,6 +249,28 @@ public static DeleteFile writePosDeleteFile(
return posWriter.toDeleteFile();
}
+ public static DeleteFile writePosDeleteFile(
+ Table table,
+ PartitionSpec spec,
+ String filename,
+ FileWriterFactory<RowData> writerFactory,
+ List<Pair<CharSequence, Long>> positions)
+ throws IOException {
+ EncryptedOutputFile outputFile =
+ encrypt(fromPath(new Path(table.location(), filename), new Configuration()));
+
+ PositionDeleteWriter<RowData> posWriter =
+ writerFactory.newPositionDeleteWriter(outputFile, spec, null);
+ PositionDelete<RowData> posDelete = PositionDelete.create();
+ try (PositionDeleteWriter<RowData> writer = posWriter) {
+ for (Pair<CharSequence, Long> p : positions) {
+ writer.write(posDelete.set(p.first(), p.second(), null));
Review Comment:
Good point.
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #14271:
URL: https://github.com/apache/iceberg/pull/14271
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2418072889
##########
core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java:
##########
@@ -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(
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2418069419
##########
data/src/main/java/org/apache/iceberg/data/BaseFileWriterFactory.java:
##########
@@ -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 {
Review Comment:
ah. I didn't notice that `BaseTable` implements `writeReplace`. thanks for explaining
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2428176345
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkFileWriterFactory.java:
##########
@@ -42,13 +42,14 @@
import org.apache.iceberg.orc.ORC;
import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+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 {
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2411084208
##########
data/src/main/java/org/apache/iceberg/data/BaseFileWriterFactory.java:
##########
@@ -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 {
Review Comment:
needed for Flink serialization
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2418569522
##########
core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java:
##########
@@ -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(
Review Comment:
If we deprecate the `AppenderFactory` then we can remove the if statements.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2415579978
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java:
##########
@@ -240,6 +249,28 @@ public static DeleteFile writePosDeleteFile(
return posWriter.toDeleteFile();
}
+ public static DeleteFile writePosDeleteFile(
+ Table table,
+ PartitionSpec spec,
+ String filename,
+ FileWriterFactory<RowData> writerFactory,
+ List<Pair<CharSequence, Long>> positions)
+ throws IOException {
+ EncryptedOutputFile outputFile =
+ encrypt(fromPath(new Path(table.location(), filename), new Configuration()));
+
+ PositionDeleteWriter<RowData> posWriter =
+ writerFactory.newPositionDeleteWriter(outputFile, spec, null);
+ PositionDelete<RowData> posDelete = PositionDelete.create();
+ try (PositionDeleteWriter<RowData> writer = posWriter) {
+ for (Pair<CharSequence, Long> p : positions) {
+ writer.write(posDelete.set(p.first(), p.second(), null));
Review Comment:
Do we need to use `set(CharSequence newPath, long newPos)` in this PR since the method we are currently using has been marked as deprecated?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2415121310
##########
data/src/main/java/org/apache/iceberg/data/BaseFileWriterFactory.java:
##########
@@ -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 {
Review 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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2412576484
##########
core/src/main/java/org/apache/iceberg/io/PartitionedWriter.java:
##########
@@ -46,6 +46,16 @@ protected PartitionedWriter(
super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
}
+ protected PartitionedWriter(
Review Comment:
Yeah, it is not changed with this PR.
But, yeah, we might want to deprecate it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#issuecomment-3400469311
Merged to main.
Thanks for the review @stevenzwu!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2416118196
##########
data/src/main/java/org/apache/iceberg/data/BaseFileWriterFactory.java:
##########
@@ -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 {
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2427033069
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkFileWriterFactory.java:
##########
@@ -42,13 +42,14 @@
import org.apache.iceberg.orc.ORC;
import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+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 {
Review Comment:
nit: looks like the `public` is only needed for `SimpleDataUtil` and other test class. Should we add `@Internal` and `@VisibleForTest` annotations?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#issuecomment-3378339818
CC: @stevenzwu, @Guosmilesmile, @mxm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2411249896
##########
core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java:
##########
@@ -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(
Review 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.
##########
core/src/main/java/org/apache/iceberg/io/PartitionedWriter.java:
##########
@@ -46,6 +46,16 @@ protected PartitionedWriter(
super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
}
+ protected PartitionedWriter(
Review Comment:
looks like PartitionedWriter is not used in Iceberg repo anymore
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Move write from AppenderFactory to FileWriterFactory [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14271:
URL: https://github.com/apache/iceberg/pull/14271#discussion_r2412617023
##########
core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java:
##########
@@ -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(
Review 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org