coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2049717807
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -407,15 +409,49 @@ private Builder doExecuteWithPartialProgress(
Stream<RewriteFileGroup> toGroupStream(
RewriteExecutionContext ctx, Map<StructLike, List<List<FileScanTask>>>
groupsByPartition) {
- return groupsByPartition.entrySet().stream()
+ if (maxFilesToRewrite == null) {
+ return groupsByPartition.entrySet().stream()
+ .filter(e -> !e.getValue().isEmpty())
+ .flatMap(
+ e -> {
+ StructLike partition = e.getKey();
+ List<List<FileScanTask>> scanGroups = e.getValue();
+ return scanGroups.stream().map(tasks -> newRewriteGroup(ctx,
partition, tasks));
+ })
+ .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ }
+
+ LOG.info(
+ "Max files rewrite options provided for table {} with max file
re-write value : {}",
+ table.name(),
+ maxFilesToRewrite);
+
+ List<RewriteFileGroup> selectedFileGroups = Lists.newArrayList();
+ AtomicInteger fileCountRunner = new AtomicInteger(0);
+
+ groupsByPartition.entrySet().stream()
+ .parallel()
Review Comment:
Good point ! I initially planned to do this in parallel to speed up the
compute. However I figured that I would plan to make such changes to all other
stream operations with benchmarking in near future and I reverted the parallel
call on the stream to keep things consistent. Thank you !
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2028,6 +2028,42 @@ protected List<DataFile> currentDataFiles(Table table) {
.collect(Collectors.toList());
}
+ @TestTemplate
+ public void testRewriteMaxFilesOption() {
Review Comment:
@sririshindra , Thank you for pointing this out. I developed this code over
internal mirror which broke tests once I rebased to the latest main changes. I
was able to verify on my local )through IntelliJ and gradle) that the tests
(both the ones I added and the existing ones) do pass.
--
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
