eqd-1 final

Author: Babar

pyiceberg/io/pyarrow.py

@@ -2757,23 +2757,20 @@ def _apply_equality_deletes(data_table: pa.Table, delete_table: pa.Table, equali
     for col in equality_columns:
         data_type = data_table.schema.field(col).type
         if delete_table[col].type != data_type:
-            delete_type_matched = delete_type_matched.set_column( delete_type_matched.schema.get_field_index(col), col, delete_table[col].cast(data_type) )
+            delete_type_matched = delete_type_matched.set_column(
+                delete_type_matched.schema.get_field_index(col),
+                col,
+                delete_table[col].cast(data_type)
+            )
 
     delete_keys = set()
     for row in delete_type_matched.to_pylist():
         key = tuple(row[col] for col in equality_columns)
         delete_keys.add(key)
 
+    column_arrays = [data_table[col] for col in equality_columns]
 
-    mask = []
-    for i in range(len(data_table)):
-        row_values = []
-        for col in equality_columns:
-            value = data_table[col][i].as_py()
-            row_values.append(value)
-
-        row_tuple = tuple(row_values)
-        mask.append(row_tuple not in delete_keys)
+    mask = [(tuple(arr[i].as_py() for arr in column_arrays) not in delete_keys) for i in range(len(data_table))]
 
     result = data_table.filter(pa.array(mask))
     return result

pyiceberg/table/__init__.py

@@ -1691,7 +1691,7 @@ def _match_equality_deletes_to_data_file(data_entry: ManifestEntry, equality_del
     relevant_entries = []
     for entry in equality_delete_entries:
         if entry.data_file.file_path == data_entry.data_file.file_path:
-            if (entry.sequence_number or INITIAL_SEQUENCE_NUMBER) < (data_entry.sequence_number or INITIAL_SEQUENCE_NUMBER):
+            if (entry.sequence_number - 1 or INITIAL_SEQUENCE_NUMBER) >= (data_entry.sequence_number or INITIAL_SEQUENCE_NUMBER):
                 if (entry.data_file.partition == data_entry.data_file.partition and entry.data_file.spec_id == data_entry.data_file.spec_id) or (entry.data_file.spec_id == UNPARTITIONED_PARTITION_SPEC.spec_id):
                     if entry.data_file.equality_ids is not None and data_entry.data_file.equality_ids is not None and set(entry.data_file.equality_ids) == set(data_entry.data_file.equality_ids):
                         relevant_entries.append(entry)