[IcebergIO] Apply delete/update filter when reading (#34607)

* apply delete filter

* revert cdc changes

* spotless

* trigger integration tests

Author: ahmedabu98

.github/trigger_files/IO_Iceberg_Integration_Tests.json

@@ -1,4 +1,4 @@
 {
     "comment": "Modify this file in a trivial way to cause this test suite to run.",
-    "modification": 4
+    "modification": 3
 }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java

@@ -32,6 +32,7 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericDeleteFilter;
 import org.apache.iceberg.data.IdentityPartitionConverters;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.data.avro.DataReader;
@@ -175,7 +176,9 @@ public boolean advance() throws IOException {
         default:
           throw new UnsupportedOperationException("Cannot read format: " + file.format());
       }
-      currentIterator = iterable.iterator();
+      GenericDeleteFilter deleteFilter =
+          new GenericDeleteFilter(checkStateNotNull(io), fileTask, fileTask.schema(), project);
+      currentIterator = deleteFilter.filter(iterable).iterator();
 
     } while (true);