You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "rdblue (via GitHub)" <gi...@apache.org> on 2023/05/28 23:43:33 UTC
[GitHub] [iceberg] rdblue opened a new pull request, #7731: Core: Implement adaptive split planning in core.
rdblue opened a new pull request, #7731:
URL: https://github.com/apache/iceberg/pull/7731
This adds utility methods for adaptive split planning to `TableScanUtil` in core.
The adaptive size is determined by eagerly loading file tasks up to the target parallelism * the requested split size. If that size is reached, then the requested split size is used. Otherwise, all files in the scan were loaded and the total size of the scan is used to create the split size by dividing the total size by the parallelism. If the resulting split size is too low, a minimum of 16MB is applied.
This differs from #7688 because that PR uses the total table size rather than a size specific to the scan. It also differs from #7714 because this is in core and is not specific to Spark. Any engine could set the scan parallelism and use adaptive split planning.
--
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
[GitHub] [iceberg] rdblue commented on pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#issuecomment-1688849081
Closing this since #7714 went in.
--
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
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1217434804
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
Can we move this logic into a separate utility method? I think we will use that directly in Spark as we have a fully loaded list of tasks and there is no point in constructing `CloseableIterable` and complicating the logic in that case.
I think this approach would perform well for cases where the provided split size prevents us from utilizing all cores in the cluster. However, I'd also target the following two use cases (I am still thinking about the second one):
- The size of data to be scanned is large and the default split size produces too many tasks. That's the original issue the PR with a SQL property was trying to solve. The need to keep all that extra task info can easily kill the driver, not to mention the overhead of launching a task in Spark (easily a second).
- I think our goal is to produce a number of splits that is a multiple of the number of slots in the cluster. Suppose we produce 1100 tasks for 1000 cores that are all equally balanced. Spark will run first 1000 tasks in parallel (let's assume each task would take the same amount of time) and only after that schedule the remaining 100 tasks. The remaining 100 tasks would take the same amount of time as the first 1000 as most of the slots in the cluster will be idle. Instead of blindly following the configured split size, it would be much better to increase the split size by 10% and process all data in one pass. That can improve the read performance up to 50%.
--
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
[GitHub] [iceberg] rdblue closed pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue closed pull request #7731: Core: Implement adaptive split planning in core.
URL: https://github.com/apache/iceberg/pull/7731
--
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
[GitHub] [iceberg] rdblue commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1220263377
##########
api/src/main/java/org/apache/iceberg/io/CloseableIterable.java:
##########
@@ -206,6 +206,10 @@ public O next() {
};
}
+ static <E> CloseableIterable<E> concat(CloseableIterable<E> i1, CloseableIterable<E> i2) {
+ return concat(ImmutableList.of(i1, i2));
Review Comment:
There was a warning about heap pollution from the parameterized type. I forgot whether we care about that so I just did this until someone can look at 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
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1220356919
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
Generating 2x by default seems reasonable to me to account for cases when certain tasks finish faster. Whether to use 1x or 2x or 4x would depend on a particular workload. I feel it is important to produce a multiple of the slots in the cluster, however. I am not sure there is a point to aim for 1.5 x splits, for instance. That's what I tried in #7714.
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
Generating 2x by default seems reasonable to me to account for cases when certain tasks finish faster. Whether to use 1x or 2x or 4x would depend on a particular workload. I feel it is important to produce a multiple of the slots in the cluster, however. I am not sure there is a point to aim for 1.5 x splits, for instance. That's what I tried to implement in #7714.
--
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
[GitHub] [iceberg] danielcweeks commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1209471374
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -35,16 +38,22 @@
import org.apache.iceberg.ScanTaskGroup;
import org.apache.iceberg.SplittableScanTask;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.io.CloseableGroup;
import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class TableScanUtil {
+ private static final Logger LOG = LoggerFactory.getLogger(TableScanUtil.class);
+ private static final long MIN_SPLIT_SIZE = 16 * 1024 * 1024; // 16 MB
Review Comment:
This feels like it's too large for a min split size. Either that or we should take into account what the row group size is actually configured to. For example, if the row group was set to 8MB, then we still wouldn't be able to achieve maximum parallelism because of this arbitrary default.
--
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
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1220201643
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
I've updated #7714 a bit to cover the two points I mentioned above. I feel using an iterator is a nice solution for regular planning when we don't have a list of tasks. However, it seems redundant for Spark as we keep a list of tasks in memory. I think approached can co-exist.
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
I've updated #7714 a bit to cover the two points I mentioned above. I feel using an iterator is a nice solution for regular planning when we don't have a list of tasks. However, it seems redundant for Spark as we keep a list of tasks in memory. I think both approaches can co-exist.
--
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
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1218711040
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -35,16 +38,22 @@
import org.apache.iceberg.ScanTaskGroup;
import org.apache.iceberg.SplittableScanTask;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.io.CloseableGroup;
import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class TableScanUtil {
+ private static final Logger LOG = LoggerFactory.getLogger(TableScanUtil.class);
+ private static final long MIN_SPLIT_SIZE = 16 * 1024 * 1024; // 16 MB
Review Comment:
+1 on 16 MB min size
--
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
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1220356919
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
Generating 2x by default seems reasonable to me to account for cases when certain tasks finish faster. Whether to use 1x or 2x or 4x would depend on a particular workload. I feel it is important to produce a multiple of the slots in the cluster, however. I am not sure there is a good use case where aiming for 1.5x slots make sense, for instance. That's what I tried to implement in #7714.
--
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
[GitHub] [iceberg] danielcweeks commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1209470679
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -79,6 +88,44 @@ public static CloseableIterable<FileScanTask> splitFiles(
return CloseableIterable.combine(splitTasks, tasks);
}
+ /**
+ * Produces {@link CombinedScanTask combined tasks} from an iterable of {@link FileScanTask file
+ * tasks}, using an adaptive target split size that targets a minimum number of tasks
+ * (parallelism).
+ *
+ * @param files incoming iterable of file tasks
+ * @param parallelism target minimum number of tasks
+ * @param splitSize target split size
+ * @param lookback bin packing lookback
+ * @param openFileCost minimum file cost
+ * @return an iterable of combined tasks
+ */
+ public static CloseableIterable<CombinedScanTask> planTasksAdaptive(
+ CloseableIterable<FileScanTask> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ long openFileCost) {
+
+ validatePlanningArguments(splitSize, lookback, openFileCost);
+
+ Function<FileScanTask, Long> weightFunc =
+ file ->
+ Math.max(
+ file.length()
+ + file.deletes().stream().mapToLong(ContentFile::fileSizeInBytes).sum(),
+ (1 + file.deletes().size()) * openFileCost);
Review Comment:
This feels like it's too large for a min split size. Either that or we should take into account what the row group size is actually configured to. For example, if the row group was set to 8MB, then we still wouldn't be able to achieve maximum parallelism because of this arbitrary default.
--
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
[GitHub] [iceberg] rdblue commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1209479290
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -79,6 +88,44 @@ public static CloseableIterable<FileScanTask> splitFiles(
return CloseableIterable.combine(splitTasks, tasks);
}
+ /**
+ * Produces {@link CombinedScanTask combined tasks} from an iterable of {@link FileScanTask file
+ * tasks}, using an adaptive target split size that targets a minimum number of tasks
+ * (parallelism).
+ *
+ * @param files incoming iterable of file tasks
+ * @param parallelism target minimum number of tasks
+ * @param splitSize target split size
+ * @param lookback bin packing lookback
+ * @param openFileCost minimum file cost
+ * @return an iterable of combined tasks
+ */
+ public static CloseableIterable<CombinedScanTask> planTasksAdaptive(
+ CloseableIterable<FileScanTask> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ long openFileCost) {
+
+ validatePlanningArguments(splitSize, lookback, openFileCost);
+
+ Function<FileScanTask, Long> weightFunc =
+ file ->
+ Math.max(
Review Comment:
This is the way it was before, but I also flagged this as something to look into.
I think we should be doing the `Math.max(fileSize, openCost)` for every file individually. Otherwise we could overpack small delete files because they aren't large compared to the size of data 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
[GitHub] [iceberg] danielcweeks commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1210586239
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -79,6 +88,44 @@ public static CloseableIterable<FileScanTask> splitFiles(
return CloseableIterable.combine(splitTasks, tasks);
}
+ /**
+ * Produces {@link CombinedScanTask combined tasks} from an iterable of {@link FileScanTask file
+ * tasks}, using an adaptive target split size that targets a minimum number of tasks
+ * (parallelism).
+ *
+ * @param files incoming iterable of file tasks
+ * @param parallelism target minimum number of tasks
+ * @param splitSize target split size
+ * @param lookback bin packing lookback
+ * @param openFileCost minimum file cost
+ * @return an iterable of combined tasks
+ */
+ public static CloseableIterable<CombinedScanTask> planTasksAdaptive(
+ CloseableIterable<FileScanTask> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ long openFileCost) {
+
+ validatePlanningArguments(splitSize, lookback, openFileCost);
+
+ Function<FileScanTask, Long> weightFunc =
+ file ->
+ Math.max(
Review Comment:
I was looking at bin packing and it seems like we would always need similar weights (e.g. you can't use file size for one task and open cost for another because they're too dissimilar). If open cost is in bytes, it feels like this should just be `totalFileSize + (fileCount * openCost)`.
--
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
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1217415171
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -79,6 +88,44 @@ public static CloseableIterable<FileScanTask> splitFiles(
return CloseableIterable.combine(splitTasks, tasks);
}
+ /**
+ * Produces {@link CombinedScanTask combined tasks} from an iterable of {@link FileScanTask file
+ * tasks}, using an adaptive target split size that targets a minimum number of tasks
+ * (parallelism).
+ *
+ * @param files incoming iterable of file tasks
+ * @param parallelism target minimum number of tasks
+ * @param splitSize target split size
+ * @param lookback bin packing lookback
+ * @param openFileCost minimum file cost
+ * @return an iterable of combined tasks
+ */
+ public static CloseableIterable<CombinedScanTask> planTasksAdaptive(
+ CloseableIterable<FileScanTask> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ long openFileCost) {
+
+ validatePlanningArguments(splitSize, lookback, openFileCost);
+
+ Function<FileScanTask, Long> weightFunc =
+ file ->
+ Math.max(
Review Comment:
The purpose of the open file cost was to avoid packing lots of tiny KB-sized files into a single split of 128 MB, which would create straggler tasks with many files. When we added this, we only had data files and we thought the cost of opening files should not matter if file sizes are reasonable.
I still tend to believe the cost of opening files should not matter for reasonably sized files to avoid changes in the way splits are packed right now. However, I can be convinced otherwise.
Comparing `fileSize` and `openCost` per each file seems worth exploring.
--
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
[GitHub] [iceberg] rdblue commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1208697877
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
This is where the split size is determined.
--
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
[GitHub] [iceberg] rdblue commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1208698377
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -79,6 +88,44 @@ public static CloseableIterable<FileScanTask> splitFiles(
return CloseableIterable.combine(splitTasks, tasks);
}
+ /**
+ * Produces {@link CombinedScanTask combined tasks} from an iterable of {@link FileScanTask file
+ * tasks}, using an adaptive target split size that targets a minimum number of tasks
+ * (parallelism).
+ *
+ * @param files incoming iterable of file tasks
+ * @param parallelism target minimum number of tasks
+ * @param splitSize target split size
+ * @param lookback bin packing lookback
+ * @param openFileCost minimum file cost
+ * @return an iterable of combined tasks
+ */
+ public static CloseableIterable<CombinedScanTask> planTasksAdaptive(
Review Comment:
If we like this approach, the new "adaptive" planning methods would be called in the scan code instead of splitting and combining. That would require adding the parallelism option to our scan API, so I'd do it only for the new `BatchScan` API.
--
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
[GitHub] [iceberg] danielcweeks commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1209478374
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
I assume your intent is that the `splitSize` would be determined before calling adaptive planning to resolve whether it's configured at the table level or the split default. The only concern is that we still need to incorporate other configurations on the table (like row group / block size) since that directly affects how we can split.
--
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
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1217415171
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -79,6 +88,44 @@ public static CloseableIterable<FileScanTask> splitFiles(
return CloseableIterable.combine(splitTasks, tasks);
}
+ /**
+ * Produces {@link CombinedScanTask combined tasks} from an iterable of {@link FileScanTask file
+ * tasks}, using an adaptive target split size that targets a minimum number of tasks
+ * (parallelism).
+ *
+ * @param files incoming iterable of file tasks
+ * @param parallelism target minimum number of tasks
+ * @param splitSize target split size
+ * @param lookback bin packing lookback
+ * @param openFileCost minimum file cost
+ * @return an iterable of combined tasks
+ */
+ public static CloseableIterable<CombinedScanTask> planTasksAdaptive(
+ CloseableIterable<FileScanTask> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ long openFileCost) {
+
+ validatePlanningArguments(splitSize, lookback, openFileCost);
+
+ Function<FileScanTask, Long> weightFunc =
+ file ->
+ Math.max(
Review Comment:
The purpose of the open file cost was to avoid packing lots of tiny KB-sized files into a single split of 128 MB, which would create straggler tasks with many files. When we added this, we only had data files and we thought the cost of opening files should not matter if file sizes are reasonable.
I still tend to believe the cost of opening files should not matter for reasonably sized files to avoid changes in the way splits are packed right now. Comparing `fileSize` and `openCost` per each file seems worth exploring.
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -79,6 +88,44 @@ public static CloseableIterable<FileScanTask> splitFiles(
return CloseableIterable.combine(splitTasks, tasks);
}
+ /**
+ * Produces {@link CombinedScanTask combined tasks} from an iterable of {@link FileScanTask file
+ * tasks}, using an adaptive target split size that targets a minimum number of tasks
+ * (parallelism).
+ *
+ * @param files incoming iterable of file tasks
+ * @param parallelism target minimum number of tasks
+ * @param splitSize target split size
+ * @param lookback bin packing lookback
+ * @param openFileCost minimum file cost
+ * @return an iterable of combined tasks
+ */
+ public static CloseableIterable<CombinedScanTask> planTasksAdaptive(
+ CloseableIterable<FileScanTask> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ long openFileCost) {
+
+ validatePlanningArguments(splitSize, lookback, openFileCost);
+
+ Function<FileScanTask, Long> weightFunc =
+ file ->
+ Math.max(
Review Comment:
The purpose of the open file cost was to avoid packing lots of tiny KB-sized files into a single split of 128 MB, which would create straggler tasks with many files. When we added this, we only had data files and we thought the cost of opening files should not matter if file sizes are reasonable.
I still tend to believe the cost of opening files should not matter for reasonably sized files to avoid changes in the way splits are packed right now.
Comparing `fileSize` and `openCost` per each file seems worth exploring.
--
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
[GitHub] [iceberg] rdblue commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1220272233
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
> Suppose we produce 1100 tasks for 1000 cores that are all equally balanced. Spark will run first 1000 tasks in parallel (let's assume each task would take the same amount of time) and only after that schedule the remaining 100 tasks. The remaining 100 tasks would take the same amount of time as the first 1000 as most of the slots in the cluster will be idle. Instead of blindly following the configured split size, it would be much better to increase the split size by 10% and process all data in one pass.
Would it be better? I've typically recommended targeting splits so that you have 2-4x the number of splits as the number of cores. That way if there is a difference in actual task runtime you can still balance the workload. For example, if you're reading 100MB splits where some of them produce 250MB of shuffle data and some produce 2.5MB of shuffle data, you want more splits to balance the workload as executors become idle.
--
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
[GitHub] [iceberg] danielcweeks commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1210583681
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -35,16 +38,22 @@
import org.apache.iceberg.ScanTaskGroup;
import org.apache.iceberg.SplittableScanTask;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.io.CloseableGroup;
import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class TableScanUtil {
+ private static final Logger LOG = LoggerFactory.getLogger(TableScanUtil.class);
+ private static final long MIN_SPLIT_SIZE = 16 * 1024 * 1024; // 16 MB
Review Comment:
I don't think we want to focus too much on optimization/cost of single tasks here. The goal is to achieve better overall wall-clock performance through parallelism and sacrifice some individual task efficiency for overall throughput. I would go as low as conceivably possible, which means either a value like this or the row group size if configured since that's the lowest splittable unit of work.
--
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
[GitHub] [iceberg] rdblue commented on pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#issuecomment-1566308612
@danielcweeks and @aokolnychyi, could you take a look and see if you like this approach?
Anton, note that this doesn't change split planning when grouping.
--
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
[GitHub] [iceberg] rdblue commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1220268210
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -79,6 +88,44 @@ public static CloseableIterable<FileScanTask> splitFiles(
return CloseableIterable.combine(splitTasks, tasks);
}
+ /**
+ * Produces {@link CombinedScanTask combined tasks} from an iterable of {@link FileScanTask file
+ * tasks}, using an adaptive target split size that targets a minimum number of tasks
+ * (parallelism).
+ *
+ * @param files incoming iterable of file tasks
+ * @param parallelism target minimum number of tasks
+ * @param splitSize target split size
+ * @param lookback bin packing lookback
+ * @param openFileCost minimum file cost
+ * @return an iterable of combined tasks
+ */
+ public static CloseableIterable<CombinedScanTask> planTasksAdaptive(
+ CloseableIterable<FileScanTask> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ long openFileCost) {
+
+ validatePlanningArguments(splitSize, lookback, openFileCost);
+
+ Function<FileScanTask, Long> weightFunc =
+ file ->
+ Math.max(
Review Comment:
> Comparing fileSize and openCost per each file seems worth exploring.
This seems more correct to me. Otherwise, a single large file could be combined with lots of small files and delete files that take too long. This is for a separate PR 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
[GitHub] [iceberg] danielcweeks commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1209476557
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -79,6 +88,44 @@ public static CloseableIterable<FileScanTask> splitFiles(
return CloseableIterable.combine(splitTasks, tasks);
}
+ /**
+ * Produces {@link CombinedScanTask combined tasks} from an iterable of {@link FileScanTask file
+ * tasks}, using an adaptive target split size that targets a minimum number of tasks
+ * (parallelism).
+ *
+ * @param files incoming iterable of file tasks
+ * @param parallelism target minimum number of tasks
+ * @param splitSize target split size
+ * @param lookback bin packing lookback
+ * @param openFileCost minimum file cost
+ * @return an iterable of combined tasks
+ */
+ public static CloseableIterable<CombinedScanTask> planTasksAdaptive(
+ CloseableIterable<FileScanTask> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ long openFileCost) {
+
+ validatePlanningArguments(splitSize, lookback, openFileCost);
+
+ Function<FileScanTask, Long> weightFunc =
+ file ->
+ Math.max(
Review Comment:
I'm a little confused by the way we're calculating this weight function. It looks like we're taking the larger of:
1. the total file size (including deletes)
2. the total number of files * open cost
Shouldn't the the weight of the task be the total file size plus the open cost for each file? With 1. above we're not including the open cost and in 2. we're not including the data to be read.
--
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
[GitHub] [iceberg] rdblue commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1209481290
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -35,16 +38,22 @@
import org.apache.iceberg.ScanTaskGroup;
import org.apache.iceberg.SplittableScanTask;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.io.CloseableGroup;
import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class TableScanUtil {
+ private static final Logger LOG = LoggerFactory.getLogger(TableScanUtil.class);
+ private static final long MIN_SPLIT_SIZE = 16 * 1024 * 1024; // 16 MB
Review Comment:
What do you think it should be instead? 8 MB?
This seems pretty reasonable to me. My rationale is that this min size should balance task overhead. We want to process enough data that we don't waste time in overhead, but not so much that we have a delay because tasks are too large. 16 MB will be processed fairly quickly and I doubt that it would be a very noticeable bottleneck. On the other hand, our file open cost is 4 MB. I would say that we want to be higher than the file open cost since that's a rough measure of overhead. So 8 or 16 seems about right to me.
--
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
[GitHub] [iceberg] rdblue commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1209536460
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
This works by targeting the split size that was requested, but trying to maximize parallelism if there isn't enough data to reach that split size. The logic for the requested split size is the same as what we currently use.
The row groups size is still orthogonal to this. We don't need to change the split size based on the target row group size. That will happen automatically because we track the split offsets and will just not be able to split files any further than their existing row groups.
--
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
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1218712116
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
> it would be much better to increase the split size by 10% and process all data in one pass. That can improve the read performance up to 50%.
Maybe the look-ahead can scan the `totalSize` up to 2x of `readToSize`. that might help to make this decision?
But this batch can always be smaller than the parallelism. we also need change in `planTasksInternal`?
--
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
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1220356919
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
Generating 2x by default seems reasonable to me to account for cases when certain tasks finish faster. Whether to use 1x or 2x or 4x would depend on a particular workload. I feel it is important to produce a multiple of the slots in the cluster, however. I am not sure there is a good use case where aiming for 1.5x slots make sense, for instance. That's what I tried to implement in #7714. That PR contains a utility method which I think can be common between these two implementation.
--
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
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1217404778
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -35,16 +38,22 @@
import org.apache.iceberg.ScanTaskGroup;
import org.apache.iceberg.SplittableScanTask;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.io.CloseableGroup;
import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class TableScanUtil {
+ private static final Logger LOG = LoggerFactory.getLogger(TableScanUtil.class);
+ private static final long MIN_SPLIT_SIZE = 16 * 1024 * 1024; // 16 MB
Review Comment:
I would say 8 MB is already small enough so such tasks should proceed fairly quickly. I would be a bit concerned going smaller than that. The cost of opening files is non-trivial. Also, too small splits may overload the underlying storage with a large number of requests. I even find 16 MB reasonable, to be honest. I don't want our queries to fail with rate limits.
--
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
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1218710601
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ Deque<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
+
+ CloseableIterable<T> tasksToReplay = CloseableIterable.withNoopClose(readAheadTasks);
+ CloseableIterable<T> allTasks;
+ if (!tasksIter.hasNext()) {
+ allTasks = tasksToReplay;
+
+ } else {
+ CloseableIterable<T> remainingTasks =
+ new CloseableIterable<T>() {
+ @Override
+ public CloseableIterator<T> iterator() {
+ return tasksIter;
+ }
+
+ @Override
+ public void close() throws IOException {
+ tasksIter.close();
Review Comment:
is this needed? `taskIter` is also closed in the finally block at line 439
--
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
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1239170345
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
I gave a bit of testing to the approach in #7714 and it led to some surprising [results](https://github.com/apache/iceberg/pull/7714#issuecomment-1603503322).
--
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
[GitHub] [iceberg] advancedxy commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1210362126
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -79,6 +88,44 @@ public static CloseableIterable<FileScanTask> splitFiles(
return CloseableIterable.combine(splitTasks, tasks);
}
+ /**
+ * Produces {@link CombinedScanTask combined tasks} from an iterable of {@link FileScanTask file
+ * tasks}, using an adaptive target split size that targets a minimum number of tasks
+ * (parallelism).
+ *
+ * @param files incoming iterable of file tasks
+ * @param parallelism target minimum number of tasks
+ * @param splitSize target split size
+ * @param lookback bin packing lookback
+ * @param openFileCost minimum file cost
+ * @return an iterable of combined tasks
+ */
+ public static CloseableIterable<CombinedScanTask> planTasksAdaptive(
+ CloseableIterable<FileScanTask> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ long openFileCost) {
+
+ validatePlanningArguments(splitSize, lookback, openFileCost);
+
+ Function<FileScanTask, Long> weightFunc =
+ file ->
+ Math.max(
+ file.length()
+ + file.deletes().stream().mapToLong(ContentFile::fileSizeInBytes).sum(),
Review Comment:
I think these two lines should be replaced as `file.sizeBytes()` like `weightFunc` in
https://github.com/apache/iceberg/pull/7731/files#diff-0b0affef002de35215ed5459cf6963fd0d5bf6d5403e67440b7ef97a35189454R170-R172
The original logic in L91-L92 in `planTasks` might not be optimal:
```
public static CloseableIterable<CombinedScanTask> planTasks(
CloseableIterable<FileScanTask> splitFiles, long splitSize, int lookback, long openFileCost) {
validatePlanningArguments(splitSize, lookback, openFileCost);
// Check the size of delete file as well to avoid unbalanced bin-packing
Function<FileScanTask, Long> weightFunc =
file ->
Math.max(
file.length()
+ file.deletes().stream().mapToLong(ContentFile::fileSizeInBytes).sum(),
(1 + file.deletes().size()) * openFileCost);
```
the `file.length` returns the the length of split/task, which would includes all the columns in one row group(for parquet). However if the table is selectively scanned, only a small portion of columns would be scanned(column pruning), the actual scanned size would be much smaller.
--
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
[GitHub] [iceberg] rdblue commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1214729399
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -79,6 +88,44 @@ public static CloseableIterable<FileScanTask> splitFiles(
return CloseableIterable.combine(splitTasks, tasks);
}
+ /**
+ * Produces {@link CombinedScanTask combined tasks} from an iterable of {@link FileScanTask file
+ * tasks}, using an adaptive target split size that targets a minimum number of tasks
+ * (parallelism).
+ *
+ * @param files incoming iterable of file tasks
+ * @param parallelism target minimum number of tasks
+ * @param splitSize target split size
+ * @param lookback bin packing lookback
+ * @param openFileCost minimum file cost
+ * @return an iterable of combined tasks
+ */
+ public static CloseableIterable<CombinedScanTask> planTasksAdaptive(
+ CloseableIterable<FileScanTask> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ long openFileCost) {
+
+ validatePlanningArguments(splitSize, lookback, openFileCost);
+
+ Function<FileScanTask, Long> weightFunc =
+ file ->
+ Math.max(
+ file.length()
+ + file.deletes().stream().mapToLong(ContentFile::fileSizeInBytes).sum(),
Review Comment:
This is a different task type that doesn't expose the same interface. The newer weight function uses a newer interface.
--
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
[GitHub] [iceberg] rdblue commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1214731215
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -35,16 +38,22 @@
import org.apache.iceberg.ScanTaskGroup;
import org.apache.iceberg.SplittableScanTask;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.io.CloseableGroup;
import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class TableScanUtil {
+ private static final Logger LOG = LoggerFactory.getLogger(TableScanUtil.class);
+ private static final long MIN_SPLIT_SIZE = 16 * 1024 * 1024; // 16 MB
Review Comment:
So you're arguing that there should be no minimum size? Seems like we might get some silly cases, like a 64k target size. That should be mostly mitigated by Parquet, but we could have a small minimum for Avro, like 4MB.
--
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
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1217434804
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
Can we move this logic into a separate utility method? I think we will use that directly in Spark as we have a fully loaded list of tasks and there is no point in constructing `CloseableIterable` and complicating the logic in that case.
I think this approach would perform well for cases where the provided split size prevents us from utilizing all cores in the cluster. However, I'd also target the following two use cases (I am still thinking about the second one):
- The size of data to be scanned is large and the default split size produces too many tasks. That's the original issue the PR with a SQL property was trying to solve. The need to keep all that extra task info can easily kill the driver, not to mention the overhead of launching a task in Spark (easily a second).
- I think our goal is to produce a number of splits that is a multiple of the number of slots in the cluster. Suppose we produce 1100 tasks for 1000 cores that are all equally balanced. Spark will run first 1000 tasks in parallel (let's assume each task would take the same amount of time) and only after that schedule the remaining 100 tasks. The remaining 100 tasks would take the same amount of time as the first 1000 as most of the slots in the cluster will be idle. Instead of blinding following the configured split size, it would be much better to increase the split size by 10% and process all data in one pass. That can improve the read performance up to 50%.
--
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
[GitHub] [iceberg] rdblue commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1220266506
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ Deque<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
+
+ CloseableIterable<T> tasksToReplay = CloseableIterable.withNoopClose(readAheadTasks);
+ CloseableIterable<T> allTasks;
+ if (!tasksIter.hasNext()) {
+ allTasks = tasksToReplay;
+
+ } else {
+ CloseableIterable<T> remainingTasks =
+ new CloseableIterable<T>() {
+ @Override
+ public CloseableIterator<T> iterator() {
+ return tasksIter;
+ }
+
+ @Override
+ public void close() throws IOException {
+ tasksIter.close();
Review Comment:
Yes, this is needed. The close on 439 closes if there is an exception or if the iterable has been exhausted (the other branch of the `tasksIter.hasNext()` check). But if an iterable is returned by this block, the responsibility to close passes to the caller and `shouldClose` is set to false on line 432.
--
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
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1220356919
##########
core/src/main/java/org/apache/iceberg/util/TableScanUtil.java:
##########
@@ -251,4 +322,126 @@ private static void validatePlanningArguments(long splitSize, int lookback, long
Preconditions.checkArgument(lookback > 0, "Split planning lookback must be > 0: %s", lookback);
Preconditions.checkArgument(openFileCost >= 0, "File open cost must be >= 0: %s", openFileCost);
}
+
+ private static <T extends ScanTask, G extends ScanTaskGroup<T>>
+ CloseableIterable<G> planTasksInternal(
+ CloseableIterable<T> splitFiles,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ Function<List<T>, G> groupFunc) {
+
+ return CloseableIterable.transform(
+ CloseableIterable.combine(
+ new BinPacking.PackingIterable<>(splitFiles, splitSize, lookback, weightFunc, true),
+ splitFiles),
+ groupFunc);
+ }
+
+ private static class AdaptiveSplitPlanningIterable<T extends ScanTask, G extends ScanTaskGroup<T>>
+ extends CloseableGroup implements CloseableIterable<G> {
+ private final CloseableIterable<T> files;
+ private final int parallelism;
+ private final long splitSize;
+ private final int lookback;
+ private final Function<T, Long> weightFunc;
+ private final BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc;
+ private final Function<List<T>, G> groupFunc;
+
+ private Long targetSize = null;
+
+ private AdaptiveSplitPlanningIterable(
+ CloseableIterable<T> files,
+ int parallelism,
+ long splitSize,
+ int lookback,
+ Function<T, Long> weightFunc,
+ BiFunction<CloseableIterable<T>, Long, CloseableIterable<T>> splitFunc,
+ Function<List<T>, G> groupFunc) {
+ this.files = files;
+ this.parallelism = parallelism;
+ this.splitSize = splitSize;
+ this.lookback = lookback;
+ this.weightFunc = weightFunc;
+ this.splitFunc = splitFunc;
+ this.groupFunc = groupFunc;
+ }
+
+ @Override
+ public CloseableIterator<G> iterator() {
+ if (targetSize != null) {
+ // target size is already known so plan with the static target size
+ CloseableIterable<T> splitTasks = splitFunc.apply(files, targetSize);
+ CloseableIterator<G> iter =
+ planTasksInternal(splitTasks, targetSize, lookback, weightFunc, groupFunc).iterator();
+ addCloseable(iter);
+ return iter;
+ }
+
+ boolean shouldClose = true;
+ CloseableIterator<T> tasksIter = files.iterator();
+ try {
+ // load tasks until the iterator is exhausted or until the total weight is enough to get the
+ // parallelism at the split size passed in.
+ LinkedList<T> readAheadTasks = Lists.newLinkedList();
+ long readToSize = parallelism * splitSize;
+ long totalSize = 0L;
+
+ while (tasksIter.hasNext()) {
+ T task = tasksIter.next();
+ readAheadTasks.addLast(task);
+ totalSize += weightFunc.apply(task);
+
+ if (totalSize > readToSize) {
+ break;
+ }
+ }
+
+ // if total size was reached, then the requested split size is used. otherwise, the iterator
+ // was exhausted and the split size will be adjusted to target parallelism with a reasonable
+ // minimum.
+ this.targetSize = Math.max(MIN_SPLIT_SIZE, Math.min(totalSize / parallelism, splitSize));
Review Comment:
Generating 2x by default seems reasonable to me to account for cases when certain tasks finish faster. Whether to use 1x or 2x or 4x would depend on a particular workload. I feel it is important to produce a multiple of the slots in the cluster, however. I am not sure there is a good use case where aiming for 1.5x slots make sense, for instance. That's what I tried to implement in #7714. That PR contains a utility method which I think can be shared by these two implementations.
--
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
[GitHub] [iceberg] ConeyLiu commented on pull request #7731: Core: Implement adaptive split planning in core.
Posted by "ConeyLiu (via GitHub)" <gi...@apache.org>.
ConeyLiu commented on PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#issuecomment-1584052810
Hi @rdblue, how to determine the target parallelism for a job such as Spark which enabled dynamic resource allocation? I think it has been set as enabled by default for many Spark application jobs in production.
--
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
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #7731: Core: Implement adaptive split planning in core.
Posted by "jackye1995 (via GitHub)" <gi...@apache.org>.
jackye1995 commented on code in PR #7731:
URL: https://github.com/apache/iceberg/pull/7731#discussion_r1214749370
##########
api/src/main/java/org/apache/iceberg/io/CloseableIterable.java:
##########
@@ -206,6 +206,10 @@ public O next() {
};
}
+ static <E> CloseableIterable<E> concat(CloseableIterable<E> i1, CloseableIterable<E> i2) {
+ return concat(ImmutableList.of(i1, i2));
Review Comment:
nit: why not use `CloseableIterable<E>...`
--
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