NikitaMatskevich commented on code in PR #13459:
URL: https://github.com/apache/iceberg/pull/13459#discussion_r2183601029
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
@@ -312,22 +316,42 @@ private String rebuildMetadata() {
}
private String saveFileList(Set<Pair<String, String>> filesToMove) {
- List<Tuple2<String, String>> fileList =
- filesToMove.stream()
- .map(p -> Tuple2.apply(p.first(), p.second()))
- .collect(Collectors.toList());
- Dataset<Tuple2<String, String>> fileListDataset =
- spark().createDataset(fileList, Encoders.tuple(Encoders.STRING(),
Encoders.STRING()));
String fileListPath = stagingDir + RESULT_LOCATION;
- fileListDataset
- .repartition(1)
- .write()
- .mode(SaveMode.Overwrite)
- .format("csv")
- .save(fileListPath);
+
+ // We need ResolvingFileIO in case we are writing to another filesystem.
+ try (FileIO io = fetchResolvingFileIO()) {
+ // Write using native Iceberg FileIO instead of relying on Spark writers
+ OutputFile fileList = io.newOutputFile(fileListPath);
+ writeAsCsv(filesToMove, fileList);
+ }
+
return fileListPath;
}
+ private void writeAsCsv(Set<Pair<String, String>> csvRows, OutputFile
outputFile) {
+ try (BufferedWriter writer = new BufferedWriter(
+ new OutputStreamWriter(outputFile.create(),
StandardCharsets.UTF_8))) {
+ for (Pair<String, String> pair : csvRows) {
+ writer.write(String.join(",", pair.first(), pair.second()));
+ writer.newLine();
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e);
+ }
+ }
+
+ private FileIO fetchResolvingFileIO() {
+ FileIO tableIO = table.io();
+ ResolvingFileIO resolvingIO = (ResolvingFileIO) CatalogUtil.loadFileIO(
Review Comment:
After looking at the code again I realized that stagingDir couldn't have
been used with another filesystem as you said. I will simplify this part. Thank
you for the review!
--
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-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
