This is an automated email from the ASF dual-hosted git repository.
huaxingao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new ad4e824433 Core: Avoid redundant SplitScanTask wrapper for files that
fit in a single split (#18002)
ad4e824433 is described below
commit ad4e824433e543affb8dd05b6c8e7f4ab1a5fd3c
Author: Yuming Wang <[email protected]>
AuthorDate: Wed Sep 23 07:08:57 2026 +0800
Core: Avoid redundant SplitScanTask wrapper for files that fit in a single
split (#18002)
* Core: Avoid redundant split wrapper for files smaller than target split
size
BaseContentScanTask.split() wraps every splittable file in a SplitScanTask
even when the file already fits within a single split (length <=
targetSplitSize).
For tables with many small files this allocates one redundant 1:1 wrapper
per data file during scan planning, adding Spark driver heap pressure.
Short-circuit that case: when a file fits in a single split, return the
task itself instead of constructing an iterator that would only ever
produce a single, functionally-identical wrapper.
Also make IcebergSourceSplit.splitId() class-agnostic so the split
identity is stable whether the underlying task is a BaseFileScanTask or
a SplitScanTask. Previously splitId embedded the concrete task class name
via MoreObjects.toStringHelper(fileScanTask), so the serialization
round-trip was not type-preserving (ScanTaskParser.fromJson always
returns a SplitScanTask), breaking TestIcebergEnumeratorStateSerializer.
Use a fixed label so splitId no longer depends on the runtime task type.
Co-authored-by: Cursor <[email protected]>
* fix
---------
Co-authored-by: Cursor <[email protected]>
---
.../org/apache/iceberg/BaseContentScanTask.java | 2 +
.../org/apache/iceberg/util/TestTableScanUtil.java | 140 ++++++++++++++++++---
2 files changed, 123 insertions(+), 19 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/BaseContentScanTask.java
b/core/src/main/java/org/apache/iceberg/BaseContentScanTask.java
index 53c2875a70..f854acffbc 100644
--- a/core/src/main/java/org/apache/iceberg/BaseContentScanTask.java
+++ b/core/src/main/java/org/apache/iceberg/BaseContentScanTask.java
@@ -104,6 +104,8 @@ abstract class BaseContentScanTask<ThisT extends
ContentScanTask<F>, F extends C
return () ->
new OffsetsAwareSplitScanTaskIterator<>(
self(), length(), splitOffsets, this::newSplitTask);
+ } else if (length() <= targetSplitSize && length() > 0) {
+ return ImmutableList.of(self());
} else {
return () ->
new FixedSizeSplitScanTaskIterator<>(
diff --git a/core/src/test/java/org/apache/iceberg/util/TestTableScanUtil.java
b/core/src/test/java/org/apache/iceberg/util/TestTableScanUtil.java
index 63db332d8c..c2caadc294 100644
--- a/core/src/test/java/org/apache/iceberg/util/TestTableScanUtil.java
+++ b/core/src/test/java/org/apache/iceberg/util/TestTableScanUtil.java
@@ -151,26 +151,9 @@ public class TestTableScanUtil {
@Test
public void testTaskGroupPlanningCorruptedOffset() {
- DataFile dataFile =
- DataFiles.builder(TestBase.SPEC)
- .withPath("/path/to/data-a.parquet")
- .withFileSizeInBytes(10)
- .withPartitionPath("data_bucket=0")
- .withRecordCount(1)
- .withSplitOffsets(
- ImmutableList.of(2L, 12L)) // the last offset is beyond the
end of the file
- .build();
-
- ResidualEvaluator residualEvaluator =
- ResidualEvaluator.of(TestBase.SPEC, Expressions.equal("id", 1), false);
-
+ // the last offset is beyond the end of the file
BaseFileScanTask baseFileScanTask =
- new BaseFileScanTask(
- dataFile,
- null,
- SchemaParser.toJson(TestBase.SCHEMA),
- PartitionSpecParser.toJson(TestBase.SPEC),
- residualEvaluator);
+ newScanTask(FileFormat.PARQUET, 10L, ImmutableList.of(2L, 12L));
List<BaseFileScanTask> baseFileScanTasks =
ImmutableList.of(baseFileScanTask);
@@ -332,6 +315,125 @@ public class TestTableScanUtil {
.hasMessageStartingWith("Parallelism must be > 0: 0");
}
+ @Test
+ public void testSplitSkipsWrapWhenFileFitsSingleSplit() {
+ // A small splittable file (no offsets) smaller than the target split size
must not be wrapped
+ // in a redundant 1:1 SplitScanTask; split() should return the task itself.
+ BaseFileScanTask task = newScanTask(FileFormat.PARQUET, 64L, null);
+
+ List<FileScanTask> splits = ImmutableList.copyOf(task.split(128L));
+
+ assertThat(splits).hasSize(1);
+ assertThat(splits.get(0)).isSameAs(task);
+ // behavior must be identical to the wrapper it replaces: whole file from
offset 0
+ assertThat(splits.get(0).start()).isEqualTo(0L);
+ assertThat(splits.get(0).length()).isEqualTo(64L);
+ assertThat(splits.get(0).estimatedRowsCount()).isEqualTo(1L);
+ }
+
+ @Test
+ public void testSplitSkipsWrapWhenFileSizeEqualsTargetSplitSize() {
+ // length() == targetSplitSize is still "fits in a single split" and must
not be wrapped
+ BaseFileScanTask task = newScanTask(FileFormat.PARQUET, 128L, null);
+
+ List<FileScanTask> splits = ImmutableList.copyOf(task.split(128L));
+
+ assertThat(splits).hasSize(1);
+ assertThat(splits.get(0)).isSameAs(task);
+ }
+
+ @Test
+ public void testSplitUsesOffsetsWhenFileFitsSingleSplit() {
+ // A splittable file with valid row-group/stripe offsets is split into one
task per offset,
+ // even when the whole file fits within the target split size. The
short-circuit must not
+ // bypass OffsetsAwareSplitScanTaskIterator, which preserves
row-group-aligned splits.
+ BaseFileScanTask task = newScanTask(FileFormat.PARQUET, 64L,
ImmutableList.of(0L, 32L));
+
+ List<FileScanTask> splits = ImmutableList.copyOf(task.split(128L));
+
+ assertThat(splits).hasSize(2);
+ assertThat(splits.get(0).start()).isEqualTo(0L);
+ assertThat(splits.get(0).length()).isEqualTo(32L);
+ assertThat(splits.get(1).start()).isEqualTo(32L);
+ assertThat(splits.get(1).length()).isEqualTo(32L);
+ }
+
+ @Test
+ public void testSplitPreservesOldBehaviorForEmptyFile() {
+ // A zero-length file is not short-circuited; it falls through to the
existing splittable
+ // path. With no split offsets, FixedSizeSplitScanTaskIterator yields zero
splits, matching
+ // the behavior before this change (hasNext() == remainingLength > 0).
+ BaseFileScanTask task = newScanTask(FileFormat.PARQUET, 0L, null);
+
+ List<FileScanTask> splits = ImmutableList.copyOf(task.split(128L));
+
+ assertThat(splits).isEmpty();
+ }
+
+ @Test
+ public void testSplitFallsThroughWhenFileExceedsTargetSplitSize() {
+ // A splittable file larger than the target split size must not be
short-circuited; it falls
+ // through to FixedSizeSplitScanTaskIterator and is split into multiple
fixed-size tasks.
+ BaseFileScanTask task = newScanTask(FileFormat.PARQUET, 256L, null);
+
+ List<FileScanTask> splits = ImmutableList.copyOf(task.split(128L));
+
+ assertThat(splits).hasSize(2);
+ assertThat(splits.get(0).start()).isEqualTo(0L);
+ assertThat(splits.get(0).length()).isEqualTo(128L);
+ assertThat(splits.get(1).start()).isEqualTo(128L);
+ assertThat(splits.get(1).length()).isEqualTo(128L);
+ }
+
+ @Test
+ public void testSplitPreservesOldBehaviorForEmptyFileWithOffsets() {
+ // A zero-length file's split offsets are not "well defined" (the last
offset 0 is not less
+ // than the file size 0), so they are ignored and the file falls through to
+ // FixedSizeSplitScanTaskIterator, yielding zero splits just like the
no-offsets empty file.
+ BaseFileScanTask task = newScanTask(FileFormat.PARQUET, 0L,
ImmutableList.of(0L));
+
+ List<FileScanTask> splits = ImmutableList.copyOf(task.split(128L));
+
+ assertThat(splits).isEmpty();
+ }
+
+ @Test
+ public void testSplitSkipsWrapWhenOffsetsNotStrictlyAscending() {
+ // Offsets that are not strictly ascending are not usable, so the file is
treated as having no
+ // usable offsets. Since it fits in a single split, split() returns the
task itself.
+ BaseFileScanTask task = newScanTask(FileFormat.PARQUET, 64L,
ImmutableList.of(32L, 0L));
+
+ List<FileScanTask> splits = ImmutableList.copyOf(task.split(128L));
+
+ assertThat(splits).hasSize(1);
+ assertThat(splits.get(0)).isSameAs(task);
+ }
+
+ private BaseFileScanTask newScanTask(
+ FileFormat format, long sizeInBytes, List<Long> splitOffsets) {
+ DataFiles.Builder builder =
+ DataFiles.builder(TestBase.SPEC)
+ .withPath("/path/to/" + format.addExtension("data-a"))
+ .withFormat(format)
+ .withFileSizeInBytes(sizeInBytes)
+ .withPartitionPath("data_bucket=0")
+ .withRecordCount(1);
+
+ if (splitOffsets != null) {
+ builder.withSplitOffsets(splitOffsets);
+ }
+
+ ResidualEvaluator residualEvaluator =
+ ResidualEvaluator.of(TestBase.SPEC, Expressions.alwaysTrue(), false);
+
+ return new BaseFileScanTask(
+ builder.build(),
+ null,
+ SchemaParser.toJson(TestBase.SCHEMA),
+ PartitionSpecParser.toJson(TestBase.SPEC),
+ residualEvaluator);
+ }
+
private PartitionScanTask taskWithPartition(
PartitionSpec spec, StructLike partition, long sizeBytes) {
PartitionScanTask task = Mockito.mock(PartitionScanTask.class);
