jackye1995 commented on a change in pull request #3171:
URL: https://github.com/apache/iceberg/pull/3171#discussion_r715304717
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
##########
@@ -538,68 +543,128 @@ protected WriterFactory(Broadcast<Table> tableBroadcast,
FileFormat format, long
@Override
public DataWriter<InternalRow> createWriter(int partitionId, long taskId,
long epochId) {
Table table = tableBroadcast.value();
-
- OutputFileFactory fileFactory = OutputFileFactory.builderFor(table,
partitionId, taskId).format(format).build();
- SparkAppenderFactory appenderFactory =
SparkAppenderFactory.builderFor(table, writeSchema, dsSchema).build();
-
PartitionSpec spec = table.spec();
FileIO io = table.io();
+ OutputFileFactory fileFactory = OutputFileFactory.builderFor(table,
partitionId, taskId)
+ .format(format)
+ .build();
+ SparkFileWriterFactory writerFactory =
SparkFileWriterFactory.builderFor(table)
+ .dataFileFormat(format)
+ .dataSchema(writeSchema)
+ .dataSparkType(dsSchema)
+ .build();
+
if (spec.isUnpartitioned()) {
- return new Unpartitioned3Writer(spec, format, appenderFactory,
fileFactory, io, targetFileSize);
+ ClusteredDataWriter<InternalRow> dataWriter = new
ClusteredDataWriter<>(
+ writerFactory, fileFactory, io,
+ format, targetFileSize);
+ return new UnpartitionedDataWriter(dataWriter, io, spec);
+
} else if (partitionedFanoutEnabled) {
- return new PartitionedFanout3Writer(
- spec, format, appenderFactory, fileFactory, io, targetFileSize,
writeSchema, dsSchema);
+ FanoutDataWriter<InternalRow> dataWriter = new FanoutDataWriter<>(
+ writerFactory, fileFactory, io,
+ format, targetFileSize);
+ return new PartitionedDataWriter(dataWriter, io, spec, writeSchema,
dsSchema);
+
} else {
- return new Partitioned3Writer(
- spec, format, appenderFactory, fileFactory, io, targetFileSize,
writeSchema, dsSchema);
+ ClusteredDataWriter<InternalRow> dataWriter = new
ClusteredDataWriter<>(
+ writerFactory, fileFactory, io,
+ format, targetFileSize);
+ return new PartitionedDataWriter(dataWriter, io, spec, writeSchema,
dsSchema);
}
}
}
- private static class Unpartitioned3Writer extends
UnpartitionedWriter<InternalRow>
- implements DataWriter<InternalRow> {
- Unpartitioned3Writer(PartitionSpec spec, FileFormat format,
SparkAppenderFactory appenderFactory,
- OutputFileFactory fileFactory, FileIO io, long
targetFileSize) {
- super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+ // TODO: why the old implementation throws throwFailureWhenFinished()?
Review comment:
It seems like we do not have consistency of this across implementations.
We did `throwFailureWhenFinished` for all the aborts in Spark, but for most of
the actions we did `suppressFailureWhenFinished`. Overall my thoughts are the
following:
1. I would prefer we keep the old behavior if possible because it's
debatable which one is more desirable, so it's probably better to just keep the
existing behavior
2. if we think suppression is better for some important reasons, we should
also log a warning `onFailure` so that at least user has a way to know what is
going on
2. ideally we can have a flag `throwOnAbortFailure` to control this from the
user side.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
