Spark 3.3, 3.4, 3.5: Remove unnecessary copying of FileScanTask (#11319)

apache · Oct 15, 2024 · 6376b44 · 6376b44
1 parent 6fdc69a
commit 6376b44
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 9 deletions.
diff --git a/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java b/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
@@ -48,7 +48,6 @@
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.spark.Spark3Util;
@@ -243,8 +242,7 @@ public boolean pushAggregation(Aggregation aggregation) {
     scan = scan.filter(filterExpression());
 
     try (CloseableIterable<FileScanTask> fileScanTasks = scan.planFiles()) {
-      List<FileScanTask> tasks = ImmutableList.copyOf(fileScanTasks);
-      for (FileScanTask task : tasks) {
+      for (FileScanTask task : fileScanTasks) {
         if (!task.deletes().isEmpty()) {
           LOG.info("Skipping aggregate pushdown: detected row level deletes");
           return false;

diff --git a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
@@ -49,7 +49,6 @@
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.metrics.InMemoryMetricsReporter;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.spark.Spark3Util;
@@ -239,8 +238,7 @@ public boolean pushAggregation(Aggregation aggregation) {
         buildIcebergBatchScan(true /* include Column Stats */, schemaWithMetadataColumns());
 
     try (CloseableIterable<FileScanTask> fileScanTasks = scan.planFiles()) {
-      List<FileScanTask> tasks = ImmutableList.copyOf(fileScanTasks);
-      for (FileScanTask task : tasks) {
+      for (FileScanTask task : fileScanTasks) {
         if (!task.deletes().isEmpty()) {
           LOG.info("Skipping aggregate pushdown: detected row level deletes");
           return false;

diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
@@ -49,7 +49,6 @@
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.metrics.InMemoryMetricsReporter;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.spark.Spark3Util;
@@ -239,8 +238,7 @@ public boolean pushAggregation(Aggregation aggregation) {
         buildIcebergBatchScan(true /* include Column Stats */, schemaWithMetadataColumns());
 
     try (CloseableIterable<FileScanTask> fileScanTasks = scan.planFiles()) {
-      List<FileScanTask> tasks = ImmutableList.copyOf(fileScanTasks);
-      for (FileScanTask task : tasks) {
+      for (FileScanTask task : fileScanTasks) {
         if (!task.deletes().isEmpty()) {
           LOG.info("Skipping aggregate pushdown: detected row level deletes");
           return false;