Core, Spark 4.1: Fix distributed planning for CoW operations (#15246)

Author: aokolnychyi

core/src/main/java/org/apache/iceberg/BaseDistributedDataScan.java

@@ -299,6 +299,10 @@ private DeleteFileIndex planDeletesLocally(List<ManifestFile> deleteManifests) {
       builder.planWith(planExecutor());
     }
 
+    if (shouldIgnoreResiduals()) {
+      builder.ignoreResiduals();
+    }
+
     return builder
         .specsById(specs())
         .filterData(filter())

spark/v4.1/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java

@@ -26,6 +26,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.ClosingIterator;
 import org.apache.iceberg.io.FileIO;
@@ -253,7 +254,7 @@ private static class ReadDeleteManifest implements FlatMapFunction<ManifestFileB
 
     ReadDeleteManifest(Broadcast<Table> table, TableScanContext context) {
       this.table = table;
-      this.filter = context.rowFilter();
+      this.filter = context.ignoreResiduals() ? Expressions.alwaysTrue() : context.rowFilter();
       this.isCaseSensitive = context.caseSensitive();
     }
 

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java

@@ -693,8 +693,7 @@ public Scan buildCopyOnWriteScan() {
     Schema expectedSchema = schemaWithMetadataColumns();
 
     BatchScan scan =
-        table
-            .newBatchScan()
+        newBatchScan()
             .useSnapshot(snapshot.snapshotId())
             .ignoreResiduals()
             .caseSensitive(caseSensitive)

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java

@@ -302,12 +302,12 @@ public void limitPushedDownToSparkScan() {
         .isEqualTo(limit);
 
     // verify CoW scan
-    assertThat(builder.buildCopyOnWriteScan())
-        .extracting("scan")
-        .extracting("scan")
-        .extracting("context")
-        .extracting("minRowsRequested")
-        .isEqualTo(limit);
+    scanAssert = assertThat(builder.buildCopyOnWriteScan()).extracting("scan");
+    if (LOCAL == planningMode) {
+      scanAssert = scanAssert.extracting("scan");
+    }
+
+    scanAssert.extracting("context").extracting("minRowsRequested").isEqualTo(limit);
 
     // verify MoR scan
     scanAssert = assertThat(builder.buildMergeOnReadScan()).extracting("scan");