You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "coderfender (via GitHub)" <gi...@apache.org> on 2025/04/17 06:22:50 UTC
[PR] [Spark]Add max files rewrite option for RewriteAction [iceberg]
coderfender opened a new pull request, #12824:
URL: https://github.com/apache/iceberg/pull/12824
This is a new option when re-writing data files (Spark Actions) to provide user the ability to limit the number of files re-written to potentially reduce file OPS . This option is named `max-files-to-rewrite` which takes a positive integer as an input, truncates the file tasks until the value is reached. In case the table has fewer files than the parameter value, all the files are processed for re-write option. A property check to ensure that no value less than 1 has also been put in place to ensure early failure.
Implementation :
toGroupStream method in RewriteDataFilesSparkAction has been refactored to truncate the list of file scan tasks (and there by files to be processed)
An atomic integer (to ensure consistency in parallel streams) called fileCountRunner is used to update counter as the groupsByPartition is processed in parallel
In case the size of entire fileScanTask list in a partition is > maxFilesToRewrite + fileCountRunner, the fileScanTask list is truncated to only add the files until the maxFilesToRewrite value is reached.
selectedFileGroups is leveraged to hold the final file groups.
Testing :
TestRewriteDataFilesAction::testRewriteMaxFilesOption is written to handle upper bound use case where the value max-files-to-rewrite > total number of files in the table
TestRewriteDataFilesAction::testRewriteMaxFilesOptionEquality is written to handle equality use case where the value max-files-to-rewrite < total number of files in the table and the resulting data files after rewrite are equal to max-files-to-rewrite
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2838760858
Sure @pvary ,the goal here is to provide users an option to limit number of files to rewrite which could potentially destabilize jobs in case the file count is super high (which is most common in large data platforms including the one I manage atm). Also, I plan to raise a similar PR on the flink side to support this option (discussed this with @mxm internally) in the future .
However if you think we are closer in terms of merging spark rewrite refactor changes to main branch soon, I can wait on this changes, rebase code (although I am more inclined towards getting this merged first) and raise PR on new main branch
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067521513
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
Actually maybe we can just come to agreement rather than further discussion, if we just
Collect by partition, Order, then Filter with Max Rows, that should fix this right?
I would still argue this isn't really starvation, because there is always work getting done, but I wouldn't want to remove the effect of the ordering on what is chosen to be compacted.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2814516625
@anuragmantri could you take a look whenever you get a chance please?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2054973916
##########
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 == 0) {
Review Comment:
Sure ! In terms of using a default / magic value ,
1. I just set the max_files_to_rewrite as an Integer as supposed to an int (to avoid magic 0's) and handle default use case as null / Or I could init to a -1)
2. Remove brnaching but make sure there are no side effects (as not to cap the number of files should the user choose not provide an option)
3. Clean up the code and not use streams at all ? (I dont believe this is would be a good ROI given @pvary 's overhaul of the Rewrite action)
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2054926473
##########
api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java:
##########
@@ -86,6 +86,12 @@ public interface RewriteDataFiles
int MAX_CONCURRENT_FILE_GROUP_REWRITES_DEFAULT = 5;
+ /**
+ * The max number of files to be rewritten. By default, this value will be 0 (To rewrite all the
+ * files)
+ */
+ String MAX_FILES_TO_REWRITE = "max-files-to-rewrite";
Review Comment:
I'm not a fan of magic 0's. Could we just not-limit the number of files if this value isn't set? (0 could also just be an unacceptable value for a user to pass in as well)
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "anuragmantri (via GitHub)" <gi...@apache.org>.
anuragmantri commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2818943233
Also @RussellSpitzer - If you have a minute.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067497937
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
I'm still struggling a bit to figure out what class of users this is.
They know they need to set max files for shorter compactions.
They aren't running compaction frequently enough to compact their entire dataset.
They also have a specific subset of files that should be prioritized over the partitions which are getting compacted but they don't know it.
The real sticking point for me is still, why wouldn't this user just compact more frequently? Would we really be solving their problem if the number of un-compacted files is still growing monotonically over time?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2083238830
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
Review Comment:
@pvary , Basically truncated the iterator returned by `scan.planFiles()` in # 276 in BinPackRewriteFilePlanner
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2055096996
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
@RussellSpitzer , wouldn't the fact that we are enforcing a maxFilesToRewrite may cause some files not be compacted even after making it past the filtering stages? Those files that did make it past the filtering stages but had to be ignored because of the maxFilesToRewrite would potentially be starved from compaction. Isn't 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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2059241409
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
Why would they be starved? Shouldn't they come up in the next compaction? If files are being added faster than they can be compacted i'm not sure it matters which files are ignored? The comparator at the bottom would give folks the ability to pick which groups are compacted first if that's really an issue.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067555906
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
1. I am not sure if we can ever call this scenario `Stuck` . Stuck essentially means that there are files which would never ever be processed . If the user knows that the max files being rewritten is 500 and they have a spike event they sure would know that there is a backlog accrued which would be processed (albeit slowly) eventually
2. Also users could perhaps set the option to be a dynamically generated option through simple scripting to make sure compaction runs on a proportionate scale ? (say max files to rewrite = % age of data files written / day ?)
3. This option is intended / encouraged to have an upper limit to the number of files to be processed in a single go to improve reliability . In my use case where we migrated existing data warehouse (un-compacted / partially compacted) , there were > 1 billion files and issuing a rewrite would essentially break the name node . Having this option would ensure files are processed at practical pace to eventually finish compaction
4. In my experience with compacting large data clusters ( ~ 25 PB) and > billion files , simple dashboards essentially help us to overlay blocks vs files to essentially optimize / tune compaction scale which I think is more on a metric layer than at the framework layer . Returning uncompacted files or (essentially total files - `max files to rewrite` is probably not very useful IMO) .
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2080967454
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
Review Comment:
What is the biggest memory hug at this point? I expect that the `FileScanTask`s with the many `DataFile`s. If that's true, then filtering at this point doesn't save too much memory. We already have a `plan` which contains a reference for them.
Suggestions:
- Accept that we don't solve the memory issue with this solution, and concentrate on decreasing the NN calls further down the execution path. In this case filtering after sorting aligns more with the user expectations, or
- Push down the filter to `planFileGroups`, or
- Make `planFileGroups` return a real iterator (seems very hard).
WDYT?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2078569878
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
Review Comment:
Yes there is a forEach functional API loop to process fileScanTasks as per the option `max-files-to-rewrite`
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067473289
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
@RussellSpitzer
Hey Russell,
Thinking more about this, I'm not sure the comparator really solves the problem I was getting at. My understanding is that the comparator kicks in *after* the `selectedFileGroups` list is already built. It seems like it helps sort the files that *will* be compacted, but it doesn't actually influence *which* files get picked for compaction in the first place.
You said, "If files are being added faster than they can be compacted I'm not sure it matters which files are ignored." I see your point, but I think it *can* matter, especially for this reason:
Folks might write data initially in a format (like Avro) that's not the best for reading. They could be counting on compaction to rewrite those files into something better (like Parquet). If new files are constantly coming in faster than we can compact, and the system doesn't prioritize older files for compaction, some of those older Avro files might just get left behind indefinitely. That could definitely cause headaches for users trying to query that older data efficiently.
Maybe it would make sense to have a separate comparator to compare or prioritize `FileScanTask`s right when we're deciding which files to compact? We could default it to something like "oldest file first" to make sure nothing gets completely starved out. It might be overthinking it, I'm not totally sure, and I'd trust your call on whether that's necessary.
--
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] [Core] Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2093469352
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java:
##########
@@ -170,6 +170,12 @@ public Builder maxFileGroupSizeBytes(long maxFileGroupSizeBytes) {
return this;
}
+ public Builder maxFilesToRewrite(Integer maxFilesToRewrite) {
Review Comment:
Thank you . I changed the type to an int instead of Integer
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2082637481
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
Review Comment:
@pvary , I pushed a commit which moved the pruning logic rught after we get fileScanTasks from the scan API . The one good thing with this is that the implementation is easier than the above approach and the other thing to note is the file scan tasks getting pruned are always guaranteed to be random (since we are pruning before grouping the partitions) . Let me know if you think this is a clearer approach than the previous one or other wise
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2048601285
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -103,7 +105,11 @@ public class RewriteDataFilesSparkAction
private boolean useStartingSequenceNumber;
private RewriteJobOrder rewriteJobOrder;
private FileRewriter<FileScanTask, DataFile> rewriter = null;
+<<<<<<< HEAD
Review Comment:
please resolve conflicts.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2049399102
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
I suspect that this can lead to starvation. Some files may always be outside this range and may never get rewritten. One situation that I can think of where this can be a problem is if the user is using rewrite operation to convert from one file format to another. For instance, user may have a streaming job that does inserts in ORC (or Avro) format. But, after rewrite they may want it to be rewritten in parquet.
If some files are always skipped (this can potentially happen if there are more small data files being created in the meantime before the next rewrite is run) their format is never changed to parquet. Maybe you should consider shuffling the files that you want to rewrite so as to reduce the chances of starvation.
Of course, user can remove the option entirely and rewrite everything, then this issue doesn't occur. But I suspect that if the user starts using this option they may not know to switch back to the original option to tackle the starvation issue.
Similarly some partitions may always be starved i.e by the time MAX_FILES_TO_REWRITE (i.e if we decide that MAX_FILES_TO_REWRITE should be applied across all partitions) limit is reached some older partitions might not be ever compacted/rewritten. So any users that might be reading some older partitions may always be stuck with the small files problem.
##########
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:
Why are we doing this in parallel? Since we are not doing the case where the maxFilesToRewrite is null in parallel, is there a reason why we should run this in parallel here?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2049351617
##########
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:
@coderfender
The test seems to actually fail when I run it on my laptop. Here the error that I am seeing. As soon as I revert your commit I am not seeing the error. In fact all the tests in the suites are failing with your commit for me. Not sure, if this is a problem with the environment in my laptop or something missing with your code. I haven't digged deep into it yet. I will do a more thorough review as soon as I get a chance.
```
Cannot downgrade v3 table to v2
java.lang.IllegalArgumentException: Cannot downgrade v3 table to v2
at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkArgument(Preconditions.java:301)
at org.apache.iceberg.TableMetadata$Builder.upgradeFormatVersion(TableMetadata.java:1070)
at org.apache.iceberg.TableMetadata.replaceProperties(TableMetadata.java:632)
at org.apache.iceberg.PropertiesUpdate.lambda$commit$0(PropertiesUpdate.java:113)
at org.apache.iceberg.util.Tasks$Builder.runTaskWithRetry(Tasks.java:413)
at org.apache.iceberg.util.Tasks$Builder.runSingleThreaded(Tasks.java:219)
at org.apache.iceberg.util.Tasks$Builder.run(Tasks.java:203)
at org.apache.iceberg.util.Tasks$Builder.run(Tasks.java:196)
at org.apache.iceberg.PropertiesUpdate.commit(PropertiesUpdate.java:110)
at org.apache.iceberg.spark.actions.TestRewriteDataFilesAction.testRewriteMaxFilesOption(TestRewriteDataFilesAction.java:2038)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at java.base/java.util.Optional.ifPresent(Optional.java:183)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
```
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2049724630
##########
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) {
Review Comment:
Good point ! My motivation here is to keep the option as simple and straightforward as possible. Having an upper bound ( I am assuming you meant setting default value of param MAX_FILES_TO_REWRITE as LONG.MAX_VALUE) which would add a side effect to this functionality by limiting the number of file to 2^^63-1 . However unlikely that is, I dont believe optional parameters should interfere in the default behavior and isolate for the sake of consistency and clarity.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2054970782
##########
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 == 0) {
Review Comment:
I realize this is getting a bit ugly so feel free to refactor this whole mess into something that doesn't use streams. We may as well just do some "breaks" and returns, or "while"'s
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2059254400
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2011,6 +2011,40 @@ public void testZOrderRewriteWithSpecificOutputSpecId() {
shouldRewriteDataFilesWithPartitionSpec(table, outputSpecId);
}
+ @TestTemplate
+ public void testRewriteMaxFilesOption() {
+ Table table = createTablePartitioned(5, 4);
+ shouldHaveFiles(table, 20);
+ int desiredMaxFileCount = 5000;
+ Result result =
+ actions()
+ .rewriteDataFiles(table)
+ .option(RewriteDataFiles.MAX_FILES_TO_REWRITE, String.valueOf(desiredMaxFileCount))
+ .execute();
+ table.refresh();
+ int filesReWritten = result.rewrittenDataFilesCount();
+ assertThat(filesReWritten).isLessThanOrEqualTo(desiredMaxFileCount);
+ }
+
+ @TestTemplate
+ public void testRewriteMaxFilesOptionEquality() {
+ Table table = createTablePartitioned(5, 4);
+ writeRecords(1, SCALE, 1);
+ writeRecords(2, SCALE, 2);
+ writeRecords(3, SCALE, 3);
+ writeRecords(4, SCALE, 4);
+ shouldHaveFiles(table, 50);
+ int desiredMaxFileCount = 10;
+ Result result =
+ actions()
+ .rewriteDataFiles(table)
+ .option(RewriteDataFiles.MAX_FILES_TO_REWRITE, "10")
+ .execute();
+ table.refresh();
+ int filesReWritten = result.rewrittenDataFilesCount();
+ assertThat(filesReWritten).isEqualTo(desiredMaxFileCount);
+ }
+
Review Comment:
I would add one test to our validation test (somewhere in this file) to test that a - or 0 value for the new param errors out as expected
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067555906
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
1. I am not sure if we can ever call this scenario `Stuck` . Stuck essentially means that there are files which would never ever be processed . If the user knows that the max files being rewritten is 500 and they have a spike event they sure would know that there is a backlog accrued which would be processed (albeit slowly) eventually
2. Also users could perhaps set the option to be a dynamically generated option through simple scripting to make sure compaction runs on a proportionate scale ? (say max files to rewrite = % age of data files written / day ?)
3. This option is intended / encouraged to have an upper limit to the number of files to be processed in a single go to improve reliability . In my use case where we migrated existing data warehouse (un-compacted / partially compacted) , there were > 1 billion files and issuing a rewrite would essentially break the name node . Having this option would ensure files are processed at practical pace to eventually finish compaction
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2054973916
##########
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 == 0) {
Review Comment:
Sure ! In terms of using a default / magic value ,
1. I just set the max_files_to_rewrite as an Integer as supposed to an int (to avoid magic 0's) and handle default use case as null / Or I could init to a -1)
2. Add no additional side effects (i.e cap the number of files should the user choose not provide an option at all to not have side effects)
3. Clean up the code and not use streams at all ? (I dont believe this is would be a good ROI given @pvary 's overhaul of the Rewrite action)
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067474756
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
I think in that use case wouldn't they want to do a targeted compaction with a filter against those older files?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2077362801
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
Review Comment:
Shall we do the filtering after the sorting?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2077364031
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFilePlanner.java:
##########
@@ -290,7 +290,8 @@ void testValidOptions() {
BinPackRewriteFilePlanner.MAX_FILE_GROUP_SIZE_BYTES,
BinPackRewriteFilePlanner.DELETE_FILE_THRESHOLD,
BinPackRewriteFilePlanner.DELETE_RATIO_THRESHOLD,
- RewriteDataFiles.REWRITE_JOB_ORDER));
+ RewriteDataFiles.REWRITE_JOB_ORDER,
+ RewriteDataFiles.MAX_FILES_TO_REWRITE));
Review Comment:
We need tests in the core to test that the filtering working as expected
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2077364769
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -1888,6 +1888,88 @@ public void testZOrderRewriteWithSpecificOutputSpecId() {
shouldRewriteDataFilesWithPartitionSpec(table, outputSpecId);
}
+ @TestTemplate
+ public void testRewriteMaxFilesOption() {
Review Comment:
We need this test in core. As the feature is moved to core
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2078570233
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -19,11 +19,13 @@
package org.apache.iceberg.actions;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
-import java.util.stream.Stream;
+import org.apache.curator.shaded.com.google.common.collect.Lists;
Review Comment:
Thank you for pointing this out. I fixed this to the right Lists import
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2085140996
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
Review Comment:
Thank you for the feedback @pvary . The main goal here is to optimize the `rewrite` action itself. So in essence we want users to compact data at a volume which is more practical (which is more or less a combination of number of files and the size of the load). The option might not solve all the tricks at once but it is a potential step in the right direction .
You are right about the fact that we need to pick our poison so to speak in this usecase . Either we truncate the file list right after we get the total number of files returned after the scan and clearly mention to the users that the `max files to rewrite` are indeed the upper bound and IMHO users should be okay with that . Now, we could revert to filter the number of files per group right after the other filters / ops are applied but this might create the plan too big as per your valid concern.
I would still be inclined to go ahead and provide this option to the users and build on it in future releases
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2084484027
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
Review Comment:
I think it is not a good place to truncate.
If there are many files in the table and the first `max-files-to-rewrite` files are correctly sized, then we truncate at the given number, and later remove the other files from the plan. So basically we end skipping the compaction.
We minimally have to do the filtering after the applying the `filterFiles`, but that could still result in filtering out whole groups by `filterFileGroups`. That would cause the similar issue but less frequently. We can end up in a situation where we compact fewer files than configured.
Checked the code, and it seems that we need to filter the result of the `SizeBasedFileRewritePlanner.planFileGroups` to get the correct number of files. At this point the tasks are all materialized, so we are back in the original situation. We need to accept that we don't solve the memory issue with this solution.
WDYT? Did I miss something?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2059258772
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -407,15 +409,37 @@ private Builder doExecuteWithPartialProgress(
Stream<RewriteFileGroup> toGroupStream(
RewriteExecutionContext ctx, Map<StructLike, List<List<FileScanTask>>> groupsByPartition) {
- return groupsByPartition.entrySet().stream()
+
+ List<RewriteFileGroup> selectedFileGroups = Lists.newArrayList();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+
+ 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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (maxFilesToRewrite == null) {
+ selectedFileGroups.add(newRewriteGroup(ctx, partition, fileScanTasks));
+ } else if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
+ int fileScanTaskListSize = fileScanTasks.size();
+ selectedFileGroups.add(
+ newRewriteGroup(
+ ctx,
+ partition,
+ remainingSize >= fileScanTaskListSize
+ ? fileScanTasks
+ : fileScanTasks.subList(0, remainingSize)));
+ fileCountRunner.getAndAdd(Math.min(remainingSize, fileScanTaskListSize));
Review Comment:
I think we can avoid the terinary operator above and some of the complexity here by just calculating this Min above, and always calling subList(0, minValue)
Do we have a perf concern?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2059258772
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -407,15 +409,37 @@ private Builder doExecuteWithPartialProgress(
Stream<RewriteFileGroup> toGroupStream(
RewriteExecutionContext ctx, Map<StructLike, List<List<FileScanTask>>> groupsByPartition) {
- return groupsByPartition.entrySet().stream()
+
+ List<RewriteFileGroup> selectedFileGroups = Lists.newArrayList();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+
+ 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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (maxFilesToRewrite == null) {
+ selectedFileGroups.add(newRewriteGroup(ctx, partition, fileScanTasks));
+ } else if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
+ int fileScanTaskListSize = fileScanTasks.size();
+ selectedFileGroups.add(
+ newRewriteGroup(
+ ctx,
+ partition,
+ remainingSize >= fileScanTaskListSize
+ ? fileScanTasks
+ : fileScanTasks.subList(0, remainingSize)));
+ fileCountRunner.getAndAdd(Math.min(remainingSize, fileScanTaskListSize));
Review Comment:
I think we can avoid the terinary operator above and some of the complexity here by just calculating this Min above, and always calling subList(0, minValue)
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067487353
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
Targeted compaction is definitely an option, you're right. But the challenge is that users have to actively know or suspect that some older files aren't being compacted in the first place. Based on what I've seen, users often don't have deep insight into Iceberg's compaction planning logic. They could easily miss that older files are effectively being skipped by the standard process, meaning they might not even be aware there's a problem that targeted compaction could solve.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2077355022
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
Review Comment:
Is this run parallel somewhere? Do we need this to be `AtomicInteger`?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2813124077
Issue : https://github.com/apache/iceberg/issues/12832
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2054932871
##########
api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java:
##########
@@ -86,6 +86,12 @@ public interface RewriteDataFiles
int MAX_CONCURRENT_FILE_GROUP_REWRITES_DEFAULT = 5;
+ /**
+ * The max number of files to be rewritten. By default, this value will be 0 (To rewrite all the
+ * files)
+ */
+ String MAX_FILES_TO_REWRITE = "max-files-to-rewrite";
Review Comment:
Thank you for the feedback @RussellSpitzer ! The current logic would not limit the number of files if this optional parameter is not set (since it would never route to the else part of the logic)
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2054973916
##########
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 == 0) {
Review Comment:
Sure ! In terms of using a default / magic value ,
1. I just set the max_files_to_rewrite as an Integer as supposed to an int (to avoid magic 0's)
2. Add no additional side effects (i.e cap the number of files should the user choose not provide an option at all to not have side effects)
3. Clean up the code and not use streams at all ? (I dont believe this is would be a good ROI given @pvary 's overhaul of the Rewrite action though)
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2049351617
##########
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:
@coderfender
The test seems to actually fail when I run it on my laptop. Here is the error that I am seeing. As soon as I revert your commit I am not seeing the error. In fact all the tests in this suite are failing with your commit for me. Not sure, if this is a problem with the environment in my laptop or something missing with your code. I haven't digged deep into it yet. I will do a more thorough review as soon as I get a chance.
```
Cannot downgrade v3 table to v2
java.lang.IllegalArgumentException: Cannot downgrade v3 table to v2
at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkArgument(Preconditions.java:301)
at org.apache.iceberg.TableMetadata$Builder.upgradeFormatVersion(TableMetadata.java:1070)
at org.apache.iceberg.TableMetadata.replaceProperties(TableMetadata.java:632)
at org.apache.iceberg.PropertiesUpdate.lambda$commit$0(PropertiesUpdate.java:113)
at org.apache.iceberg.util.Tasks$Builder.runTaskWithRetry(Tasks.java:413)
at org.apache.iceberg.util.Tasks$Builder.runSingleThreaded(Tasks.java:219)
at org.apache.iceberg.util.Tasks$Builder.run(Tasks.java:203)
at org.apache.iceberg.util.Tasks$Builder.run(Tasks.java:196)
at org.apache.iceberg.PropertiesUpdate.commit(PropertiesUpdate.java:110)
at org.apache.iceberg.spark.actions.TestRewriteDataFilesAction.testRewriteMaxFilesOption(TestRewriteDataFilesAction.java:2038)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at java.base/java.util.Optional.ifPresent(Optional.java:183)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
```
--
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] [Core] Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2887503481
Sure , I will create another PR to support backport / forward port this functionality. @RussellSpitzer , @pvary
Spark - v 3.4 and 3.5 are already done so I will only have to make changes to support v4.0. Re : Flink I will have to make changes to support 1.19 and v2.0 and I am hoping all these changes can go in PR ?
--
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] [Core] Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2092711533
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java:
##########
@@ -170,6 +170,12 @@ public Builder maxFileGroupSizeBytes(long maxFileGroupSizeBytes) {
return this;
}
+ public Builder maxFilesToRewrite(Integer maxFilesToRewrite) {
Review Comment:
This is already null on the spark side of things and I chose this path to keep things consistent. I can change it to Integer instead of `int` if you think that is the right way to go @pvary ?
--
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] [Core] Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #12824:
URL: https://github.com/apache/iceberg/pull/12824
--
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] [Core] Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2887505169
I have no problem with doing all the other changes at once, I just don't like having them all in the original PR because it's harder to track changes
--
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] [Core] Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2887511976
Sure I will create a new PR just for the porting changes
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "yogevyuval (via GitHub)" <gi...@apache.org>.
yogevyuval commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2813556075
> This is a new option when re-writing data files (Spark Actions) to provide user the ability to limit the number of files re-written to potentially reduce file OPS . This option is named as `max-files-to-rewrite` which takes a positive integer as an input, truncates the file tasks until the value is reached. In case the table has fewer files than the parameter value, all the files are processed for re-write option. A property check to ensure that no value less than 1 has also been put in place to ensure early failure.
>
> Implementation :
>
> 1. `toGroupStream` method in `RewriteDataFilesSparkAction` has been refactored to truncate the list of file scan tasks (and there by files to be processed)
> 2. An atomic integer (to ensure consistency in parallel streams) called `fileCountRunner` is used to update counter as the `groupsByPartition` is processed in parallel
> 3. In case the size of entire fileScanTask list in a partition is > maxFilesToRewrite + fileCountRunner, the fileScanTask list is _truncated_ to only add the files until the maxFilesToRewrite value is reached.
> 4. `selectedFileGroups` is leveraged to hold the final file groups.
>
> Testing :
>
> 1. `TestRewriteDataFilesAction::testRewriteMaxFilesOption` is written to handle upper bound use case where the value `max-files-to-rewrite` > total number of files in the table
> 2. `TestRewriteDataFilesAction::testRewriteMaxFilesOptionEquality` is written to handle equality use case where the value `max-files-to-rewrite` < total number of files in the table and the resulting data files after rewrite are equal to `max-files-to-rewrite`
Could you elaborate on the use case that this is trying to solve? Is this to limit the resources a single job needed to reduce failures?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2054932871
##########
api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java:
##########
@@ -86,6 +86,12 @@ public interface RewriteDataFiles
int MAX_CONCURRENT_FILE_GROUP_REWRITES_DEFAULT = 5;
+ /**
+ * The max number of files to be rewritten. By default, this value will be 0 (To rewrite all the
+ * files)
+ */
+ String MAX_FILES_TO_REWRITE = "max-files-to-rewrite";
Review Comment:
Thank you for the feedback @RussellSpitzer ! Sure I can cap it to a high number such as INT_MAX (or even LONG_MAX which I dont think is a bit too much) and update documentation ?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2054958804
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
I'm not sure starvation applies here since we should only be considering files that have made it past the filtering stages. Everything here will be compacted
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2819966147
@manuzhang , Already pinged in the dev channel to get some feedback : https://apache-iceberg.slack.com/archives/C03LG1D563F/p1744871503652659 . Let me bump the message again .
Thank you for reviewing and please let me know if you see any issues with the code which I might have been missing
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2856622960
@RussellSpitzer , @pvary Rebased the branch and fixed tests. Please take a look whenever you get a chance . Thank you very much for the guidance
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067273685
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
Agreed @RussellSpitzer . I didn't realize there is a comparator available. The user can use that feature to address any starvation issues as you said.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2841286239
> Sure @pvary ,the goal here is to provide users an option to limit number of files to rewrite which could potentially destabilize jobs in case the file count is super high (which is most common in large data platforms including the one I manage atm). Also, I plan to raise a similar PR on the flink side to support this option (discussed this with @mxm internally) in the future.
My question is aimed for understanding if we really need 2 different configs. One for the number of files in the compaction plan (this PR), and another for the maximum data size for the compaction plan (feature in the Flink PR). Maybe we can merge the two configs and provide only a single configuration option which could be used for both use-cases.
In Flink's case the issue was that the planner generated too big plans, and the compaction took forever to finish when it was applied for the first time for a previously non-maintained table. Limiting the size of the data to compact in one run seemed like an intuitive option for the users. Subsequent compaction plans were ignoring the already compacted files, so the table eventually were fully compacted.
Would the size based option work for your use-case, or it needs a config around the file sizes specifically?
> However if you think we are closer in terms of merging spark rewrite refactor changes to main branch soon, I can wait on this changes, rebase code (although I am more inclined towards getting this merged first) and raise PR on new main branch
I left one single question for @RussellSpitzer, and after the answer on that we can merge the other PR immediately. (I looked my PR first, and haven't realized that the PR is such an immediate roadblock - If I knew, I might have just go with the original version). Sorry for the delay.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2054934521
##########
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 == 0) {
Review Comment:
I think it would be nice to keep this as a single branch if possible. It doesn't feel like the logic should be that different between limiting and not limiting.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2054932871
##########
api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java:
##########
@@ -86,6 +86,12 @@ public interface RewriteDataFiles
int MAX_CONCURRENT_FILE_GROUP_REWRITES_DEFAULT = 5;
+ /**
+ * The max number of files to be rewritten. By default, this value will be 0 (To rewrite all the
+ * files)
+ */
+ String MAX_FILES_TO_REWRITE = "max-files-to-rewrite";
Review Comment:
Thank you for the feedback @RussellSpitzer ! Sorry I the current logic would not limit the number of files if this optional parameter is not set (since it would never route to the else part of the logic)
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2813097314
@manuzhang , resolved conflicts and update the branch and the PR is ready for review now
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "sfc-gh-rspitzer (via GitHub)" <gi...@apache.org>.
sfc-gh-rspitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2087441627
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
Review Comment:
I think that's fair @pvary we probably can't fix the "total" memory solution but we will at least get read of a lot of object creation and make others much shorter lived.
--
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] [Core] Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2887485441
Thank you @pvary @RussellSpitzer @anuragmantri . I will start working on the documentation changes and raise a PR soon
--
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] [Core] Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2887490967
Remember to "forward port" too now that we have a 4.0 Module :)
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2818490630
@manuzhang , Could you please review this since I believe have addressed all the above issues / questions?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2838515470
@coderfender: Be aware that the planning part of the data file rewrite action is moved to core.
There is a PR under review (#12692) which will move the Spark action to use the new core API.
Also, could you please help me understand where the number of files becomes a bottleneck? We had very similar issues when working on the Flink rewrite data files. There we decide to limit the size of the full compaction (https://github.com/apache/iceberg/pull/11497/files#diff-9c2fb1c711a30af44656320fc301b42055ad4f2077c52010413eee4bf8deb854R60). We might have very similar issues here, with a slightly different solution.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2814148621
@yogevyuval , The goal here is to provide user an option to limit the number of files to be rewritten (either through compaction , data rewrite etc) . In a use case (like mine) where there are 1 billion plus files in a lake house, the user might want to iteratively run compaction to reduce the file count to an acceptable level rather than going all in at the very first time. This option should help improve rewrite spark jobs and the users can tune this param to optimize scale and reliability
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2078576476
##########
core/src/test/java/org/apache/iceberg/actions/TestBinPackRewriteFilePlanner.java:
##########
@@ -290,7 +290,8 @@ void testValidOptions() {
BinPackRewriteFilePlanner.MAX_FILE_GROUP_SIZE_BYTES,
BinPackRewriteFilePlanner.DELETE_FILE_THRESHOLD,
BinPackRewriteFilePlanner.DELETE_RATIO_THRESHOLD,
- RewriteDataFiles.REWRITE_JOB_ORDER));
+ RewriteDataFiles.REWRITE_JOB_ORDER,
+ RewriteDataFiles.MAX_FILES_TO_REWRITE));
Review Comment:
Thank you for the review . I moved tests from spark module to core `TestBinPackRewriteFilePlanner` 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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067326299
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2011,6 +2011,68 @@ public void testZOrderRewriteWithSpecificOutputSpecId() {
shouldRewriteDataFilesWithPartitionSpec(table, outputSpecId);
}
+ @TestTemplate
+ public void testRewriteMaxFilesOption() {
+ Table table = createTablePartitioned(5, 4);
+ shouldHaveFiles(table, 20);
+ int desiredMaxFileCount = 5000;
+ Result result =
+ actions()
+ .rewriteDataFiles(table)
+ .option(RewriteDataFiles.MAX_FILES_TO_REWRITE, String.valueOf(desiredMaxFileCount))
+ .execute();
+ table.refresh();
+ int filesReWritten = result.rewrittenDataFilesCount();
+ assertThat(filesReWritten).isLessThanOrEqualTo(desiredMaxFileCount);
+ }
+
+ @TestTemplate
+ public void testRewriteMaxFilesOptionEquality() {
+ Table table = createTablePartitioned(5, 4);
+ writeRecords(1, SCALE, 1);
+ writeRecords(2, SCALE, 2);
+ writeRecords(3, SCALE, 3);
+ writeRecords(4, SCALE, 4);
+ shouldHaveFiles(table, 50);
+ int desiredMaxFileCount = 10;
+ Result result =
+ actions()
+ .rewriteDataFiles(table)
+ .option(RewriteDataFiles.MAX_FILES_TO_REWRITE, "10")
+ .execute();
+ table.refresh();
+ int filesReWritten = result.rewrittenDataFilesCount();
+ assertThat(filesReWritten).isEqualTo(desiredMaxFileCount);
+ }
+
+ @TestTemplate
+ public void testRewriteMaxFilesOptionSinglePartition() {
Review Comment:
nit: This is testing testMaxFilesGreaterThanFileCount
I would also like if we can add another test with 2 partitions and max files to rewrite only truncating on one of them. Ie Partition 1 and 2 both have seven files. Should rewrite only 10 files but only take 3 of the second partition.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2054957665
##########
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 == 0) {
Review Comment:
something like
```java
long maxFilesToRewrite = Long.MAX_VALUE;
AtomicLong remainingFiles = new AtomicLong(maxFilesToRewrite);
return groupsByPartition.entrySet().stream()
.filter(e -> !e.getValue().isEmpty())
.flatMap(
e -> {
if (remainingFiles.get() == 0) {
// bail out early, no more files need to be planned for any partitions
return Stream.empty();
}
StructLike partition = e.getKey();
List<List<FileScanTask>> scanGroups = e.getValue();
return scanGroups.stream().flatMap(tasks -> {
if (remainingFiles.get() == 0) {
// bail out early, no more files need to be planned for any files within this partition
return Stream.empty();
}
int filesToTake = (int) Math.min(tasks.size(), remainingFiles.get());
remainingFiles.addAndGet(-filesToTake);
return Stream.of(newRewriteGroup(ctx, partition, tasks.subList(0, filesToTake)));
});
})
.sorted(RewriteFileGroup.comparator(rewriteJobOrder));
}
```
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067498482
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
It may help me more if you can describe a use case you would have for this?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067557994
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
@RussellSpitzer
Sorry, I didn't see your last message before sending mine!
Yes, I agree that if we "Collect by partition, Order, then Filter with Max Rows," that should indeed address the issue. Ordering (perhaps by something like file creation/arrival time, if possible) before applying the `maxFilesToRewrite` limit would prevent older files from being consistently skipped over by newer ones.
And just to quickly reiterate from my longer comment – I still think including the count of files left behind in the output would be a valuable addition to give users better visibility and help them tune their compaction strategies.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2077306640
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -19,11 +19,13 @@
package org.apache.iceberg.actions;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
-import java.util.stream.Stream;
+import org.apache.curator.shaded.com.google.common.collect.Lists;
Review Comment:
this seems off
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2078575561
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
+ .filter(e -> !e.getValue().isEmpty())
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ long inputSize = inputSize(fileScanTasks);
+ if (maxFilesToRewrite == null) {
Review Comment:
True but as per review / feature discussion from @RussellSpitzer , I had to make sure that there are absolutely no side effects for this optional parameter from a theoretical standpoint (even if it isnt practical use case to ever compact 2^^32 - 1 files).
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2082047278
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
Review Comment:
Sure the thought behind this PR was to give option to users to limit the number of files being rewritten (either due to load or due to NN pressure ). I believe that we can avoid creating a `heavy` plan object by truncating it to process by `Push down the filter to planFileGroups` is a good idea. However, do we still solve the problem of not building a large plan by moving the filter to planFileGroups method is not something I am completely clear about . We would still have to scan the table to get the files present right ?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2049439045
##########
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) {
Review Comment:
Actually can we instead set this value to LONG.MAX_FILES_TO_REWRITE by default, so that you don't have to have two code blocks to do the same thing. If the user doesn't specify this option, then MAX_FILES_TO_REWRITE is set to the default. So, along as the number of files being processed doesn't't exceed LONG.MAX_FILES_TO_REWRITE your implementation for the case when MAX_FILES_TO_REWRITE is set to some value will suffice.
If you instead decide to make the criteria something like MAX_FILES_SIZE_TO_REWRITE then the same trick works there as well.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
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-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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2049722429
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
Thank you for the review ! The motive behind this option is to provide first (of probably many) user controlled option to limit the input volume of files. Now in the above example you provided reg starvation, I am not sure if shuffle is the right way to deterministically prevent it (since the file groups itself is a result table scan and does not preserve order IMO)
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2049724630
##########
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) {
Review Comment:
Good point ! My motivation here is to keep the option as simple and straightforward as possible. Having an upper bound ( I am assuming you meant setting default value of param MAX_FILES_TO_REWRITE as LONG.MAX_VALUE) which would add a side effect to this functionality by limiting the number of file to 2^^63-1 . However unlikely that is, I dont believe optional parameters should interfere in the default behavior and should be isolated for the sake of consistency and clarity.
--
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] [Core] Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#issuecomment-2887484891
Merged to main.
Thanks for all the work @coderfender on the PR, and @RussellSpitzer for the review!
@coderfender: Could you please create the backport PRs for Spark and Flink?
Thanks, Peter
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2087432349
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
Review Comment:
Not sure if I am missing something here . @RussellSpitzer , @pvary ?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2061051874
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2011,6 +2011,40 @@ public void testZOrderRewriteWithSpecificOutputSpecId() {
shouldRewriteDataFilesWithPartitionSpec(table, outputSpecId);
}
+ @TestTemplate
+ public void testRewriteMaxFilesOption() {
+ Table table = createTablePartitioned(5, 4);
+ shouldHaveFiles(table, 20);
+ int desiredMaxFileCount = 5000;
+ Result result =
+ actions()
+ .rewriteDataFiles(table)
+ .option(RewriteDataFiles.MAX_FILES_TO_REWRITE, String.valueOf(desiredMaxFileCount))
+ .execute();
+ table.refresh();
+ int filesReWritten = result.rewrittenDataFilesCount();
+ assertThat(filesReWritten).isLessThanOrEqualTo(desiredMaxFileCount);
+ }
+
+ @TestTemplate
+ public void testRewriteMaxFilesOptionEquality() {
+ Table table = createTablePartitioned(5, 4);
+ writeRecords(1, SCALE, 1);
+ writeRecords(2, SCALE, 2);
+ writeRecords(3, SCALE, 3);
+ writeRecords(4, SCALE, 4);
+ shouldHaveFiles(table, 50);
+ int desiredMaxFileCount = 10;
+ Result result =
+ actions()
+ .rewriteDataFiles(table)
+ .option(RewriteDataFiles.MAX_FILES_TO_REWRITE, "10")
+ .execute();
+ table.refresh();
+ int filesReWritten = result.rewrittenDataFilesCount();
+ assertThat(filesReWritten).isEqualTo(desiredMaxFileCount);
+ }
+
Review Comment:
Thank you for the feedback . I added a couple more tests (to address the max files rewrite > total files in a partitions inline with your example above) and also to verify the right exception along with error message are thrown in case an invalid value is passed
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2054959143
##########
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 == 0) {
Review Comment:
Sure I initially thought of this approach but wanted to keep the code clean and isolated
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2059257480
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2011,6 +2011,40 @@ public void testZOrderRewriteWithSpecificOutputSpecId() {
shouldRewriteDataFilesWithPartitionSpec(table, outputSpecId);
}
+ @TestTemplate
+ public void testRewriteMaxFilesOption() {
+ Table table = createTablePartitioned(5, 4);
+ shouldHaveFiles(table, 20);
+ int desiredMaxFileCount = 5000;
+ Result result =
+ actions()
+ .rewriteDataFiles(table)
+ .option(RewriteDataFiles.MAX_FILES_TO_REWRITE, String.valueOf(desiredMaxFileCount))
+ .execute();
+ table.refresh();
+ int filesReWritten = result.rewrittenDataFilesCount();
+ assertThat(filesReWritten).isLessThanOrEqualTo(desiredMaxFileCount);
+ }
+
+ @TestTemplate
+ public void testRewriteMaxFilesOptionEquality() {
+ Table table = createTablePartitioned(5, 4);
+ writeRecords(1, SCALE, 1);
+ writeRecords(2, SCALE, 2);
+ writeRecords(3, SCALE, 3);
+ writeRecords(4, SCALE, 4);
+ shouldHaveFiles(table, 50);
+ int desiredMaxFileCount = 10;
+ Result result =
+ actions()
+ .rewriteDataFiles(table)
+ .option(RewriteDataFiles.MAX_FILES_TO_REWRITE, "10")
+ .execute();
+ table.refresh();
+ int filesReWritten = result.rewrittenDataFilesCount();
+ assertThat(filesReWritten).isEqualTo(desiredMaxFileCount);
+ }
+
Review Comment:
We are also missing a test where max files to rewrite causes any early termination within a partition I think. For example a table with a single partition with 7 files and max_files_to_rewrite = 10. I think the code above for subList will break here since the index I think has to be within the list itself? I haven't checked
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2061052784
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -407,15 +409,37 @@ private Builder doExecuteWithPartialProgress(
Stream<RewriteFileGroup> toGroupStream(
RewriteExecutionContext ctx, Map<StructLike, List<List<FileScanTask>>> groupsByPartition) {
- return groupsByPartition.entrySet().stream()
+
+ List<RewriteFileGroup> selectedFileGroups = Lists.newArrayList();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+
+ 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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (maxFilesToRewrite == null) {
+ selectedFileGroups.add(newRewriteGroup(ctx, partition, fileScanTasks));
+ } else if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
+ int fileScanTaskListSize = fileScanTasks.size();
+ selectedFileGroups.add(
+ newRewriteGroup(
+ ctx,
+ partition,
+ remainingSize >= fileScanTaskListSize
+ ? fileScanTasks
+ : fileScanTasks.subList(0, remainingSize)));
+ fileCountRunner.getAndAdd(Math.min(remainingSize, fileScanTaskListSize));
Review Comment:
Sure ! It was a miss from my end to assume that sublist is not O(1) complexity . I did remove the terinary operator , pre computed the variable to update the runner and also extract sublist accordingly.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2061052784
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -407,15 +409,37 @@ private Builder doExecuteWithPartialProgress(
Stream<RewriteFileGroup> toGroupStream(
RewriteExecutionContext ctx, Map<StructLike, List<List<FileScanTask>>> groupsByPartition) {
- return groupsByPartition.entrySet().stream()
+
+ List<RewriteFileGroup> selectedFileGroups = Lists.newArrayList();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+
+ 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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (maxFilesToRewrite == null) {
+ selectedFileGroups.add(newRewriteGroup(ctx, partition, fileScanTasks));
+ } else if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
+ int fileScanTaskListSize = fileScanTasks.size();
+ selectedFileGroups.add(
+ newRewriteGroup(
+ ctx,
+ partition,
+ remainingSize >= fileScanTaskListSize
+ ? fileScanTasks
+ : fileScanTasks.subList(0, remainingSize)));
+ fileCountRunner.getAndAdd(Math.min(remainingSize, fileScanTaskListSize));
Review Comment:
Sure ! It was a miss from my end to assume that sublist is not O(1) complexity . I did remove the terinary operator , pre computed the size var to update the runner and extract sublist accordingly.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067539142
##########
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()
.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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
Review Comment:
@RussellSpitzer
You're right, I might be overcomplicating the user profile a bit. Let's simplify the scenario and focus on the outcome.
Consider a user who *does* set `maxFilesToRewrite` based on their *average* daily data volume – let's say they typically write around 500 files a day and set `maxFilesToRewrite` to 700 to handle daily volume and a small buffer. They schedule their compaction to run once daily. Most days, this works fine; any slight overflow gets picked up the next day.
The issue arises during unpredictable data bursts, like a "Black Friday" event or an unexpected spike. If they suddenly get 10,000 files written on one day, their compaction job, still capped at rewriting 700 files by `maxFilesToRewrite`, will skip a massive number of newly written files (over 9,000 in this example). These skipped files form a significant backlog.
With only 700 files being compacted per day, it would take many days (potentially weeks) to clear that backlog. During this time, those thousands of files written on the burst day, and potentially many written on subsequent "normal" days while the backlog is being cleared, are simply sitting there, uncompacted. They are effectively "left behind" because the `maxFilesToRewrite` limit prevents the compaction job from even considering them in that run. The files being skipped are primarily the most recently written ones due to the limit, and they age over time while waiting to be included in a future limited compaction run. This is the scenario where older files can get "stuck" for a prolonged period if the backlog persists.
Maybe we should be thinking about giving users visibility into this backlog caused by the limit. Currently, the `rewrite_data_files` procedure gives `failed_data_files_count` as part of its output. As I understand it, this metric captures files that *failed* to be rewritten when `partial-progress.enabled` is true, but I don't believe it captures the files that were simply *skipped* because the `maxFilesToRewrite` limit was reached before they could be considered.
If the procedure's output could include a metric showing how many eligible files were *skipped* because the `maxFilesToRewrite` limit was hit, users could use this. They could monitor this skipped count. If it's consistently high or grows significantly after a data burst, they'd have clear data indicating they need to:
1. Temporarily increase `maxFilesToRewrite`.
2. Run compaction more frequently until the skipped count drops.
3. Or implement more dynamic scheduling logic based on the skipped count.
So, perhaps we need to add another parameter to the output to specifically indicate the number of files left behind due to hitting the `maxFilesToRewrite` limit. Alternatively, if `failed_data_files_count` is intended to cover this, we might need to clarify its definition in the documentation and potentially adjust the implementation and add tests to ensure it correctly reflects files skipped due to the limit, not just those that encountered rewrite errors. This enhanced output would empower users to react to unexpected volumes and prevent a long-term buildup of uncompacted files, addressing the scenario where older files remain unoptimized simply because they were part of a backlog that the fixed daily compaction couldn't clear.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2067314813
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java:
##########
@@ -407,15 +409,33 @@ private Builder doExecuteWithPartialProgress(
Stream<RewriteFileGroup> toGroupStream(
RewriteExecutionContext ctx, Map<StructLike, List<List<FileScanTask>>> groupsByPartition) {
- return groupsByPartition.entrySet().stream()
+
+ List<RewriteFileGroup> selectedFileGroups = Lists.newArrayList();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+
+ 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));
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ if (maxFilesToRewrite == null) {
+ selectedFileGroups.add(newRewriteGroup(ctx, partition, fileScanTasks));
+ } else if (fileCountRunner.get() < maxFilesToRewrite) {
+ int remainingSize = maxFilesToRewrite - fileCountRunner.get();
+ int fileScanTaskListSize = Math.min(fileScanTasks.size(), remainingSize);
Review Comment:
I would rename this I think. scanTasksToConsume? scanTasksToRewrite?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2077357196
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
+ .filter(e -> !e.getValue().isEmpty())
+ .forEach(
+ entry -> {
+ StructLike partition = entry.getKey();
+ entry
+ .getValue()
+ .forEach(
+ fileScanTasks -> {
+ long inputSize = inputSize(fileScanTasks);
+ if (maxFilesToRewrite == null) {
Review Comment:
Can we default `maxFilesToRewrite` to `Integer.MAX_VALUE` to simplify the code?
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2077932795
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
Review Comment:
I just changed my mind based on a Slack discussions yesterday. A user was hitting driver side OOMs and since we are aggregating things here in memory, it may be better to trim the amount of entities we are going to create as soon as we can.
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2077967765
##########
core/src/main/java/org/apache/iceberg/actions/BinPackRewriteFilePlanner.java:
##########
@@ -199,30 +214,48 @@ protected long defaultTargetFileSize() {
public FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan() {
StructLikeMap<List<List<FileScanTask>>> plan = planFileGroups();
RewriteExecutionContext ctx = new RewriteExecutionContext();
- Stream<RewriteFileGroup> groups =
- plan.entrySet().stream()
- .filter(e -> !e.getValue().isEmpty())
- .flatMap(
- e -> {
- StructLike partition = e.getKey();
- List<List<FileScanTask>> scanGroups = e.getValue();
- return scanGroups.stream()
- .map(
- tasks -> {
- long inputSize = inputSize(tasks);
- return newRewriteGroup(
- ctx,
- partition,
- tasks,
- inputSplitSize(inputSize),
- expectedOutputFiles(inputSize));
- });
- })
- .sorted(RewriteFileGroup.comparator(rewriteJobOrder));
+ List<RewriteFileGroup> selectedFileGroups = new ArrayList<>();
+ AtomicInteger fileCountRunner = new AtomicInteger();
+ plan.entrySet().stream()
Review Comment:
Makes sense. If not the NN is the issue, then it makes sense to limit the number of file ASAP
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2078576961
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -1888,6 +1888,88 @@ public void testZOrderRewriteWithSpecificOutputSpecId() {
shouldRewriteDataFilesWithPartitionSpec(table, outputSpecId);
}
+ @TestTemplate
+ public void testRewriteMaxFilesOption() {
Review Comment:
Great point ! I did move all the tests from spark module to `TestBinPackRewriteFilePlanner` 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] [Core] Add max files rewrite option for RewriteAction [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2092664923
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java:
##########
@@ -170,6 +170,12 @@ public Builder maxFileGroupSizeBytes(long maxFileGroupSizeBytes) {
return this;
}
+ public Builder maxFilesToRewrite(Integer maxFilesToRewrite) {
Review Comment:
Should be `int`. I don't think you want to set it to `null`
--
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] [Spark]Add max files rewrite option for RewriteAction [iceberg]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #12824:
URL: https://github.com/apache/iceberg/pull/12824#discussion_r2049351617
##########
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:
@coderfender
The test seems to actually fail when I run it on my laptop. Here is the error that I am seeing. As soon as I revert your commit I am not seeing the error. In fact all the tests in the suites are failing with your commit for me. Not sure, if this is a problem with the environment in my laptop or something missing with your code. I haven't digged deep into it yet. I will do a more thorough review as soon as I get a chance.
```
Cannot downgrade v3 table to v2
java.lang.IllegalArgumentException: Cannot downgrade v3 table to v2
at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkArgument(Preconditions.java:301)
at org.apache.iceberg.TableMetadata$Builder.upgradeFormatVersion(TableMetadata.java:1070)
at org.apache.iceberg.TableMetadata.replaceProperties(TableMetadata.java:632)
at org.apache.iceberg.PropertiesUpdate.lambda$commit$0(PropertiesUpdate.java:113)
at org.apache.iceberg.util.Tasks$Builder.runTaskWithRetry(Tasks.java:413)
at org.apache.iceberg.util.Tasks$Builder.runSingleThreaded(Tasks.java:219)
at org.apache.iceberg.util.Tasks$Builder.run(Tasks.java:203)
at org.apache.iceberg.util.Tasks$Builder.run(Tasks.java:196)
at org.apache.iceberg.PropertiesUpdate.commit(PropertiesUpdate.java:110)
at org.apache.iceberg.spark.actions.TestRewriteDataFilesAction.testRewriteMaxFilesOption(TestRewriteDataFilesAction.java:2038)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at java.base/java.util.Optional.ifPresent(Optional.java:183)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
```
--
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