[Append Scan] Integration tests for IncrementalAppendScan

dev/provision.py

@@ -409,3 +409,58 @@
     )
     spark.sql(f"ALTER TABLE {catalog_name}.default.test_empty_scan_ordered_str WRITE ORDERED BY id")
     spark.sql(f"INSERT INTO {catalog_name}.default.test_empty_scan_ordered_str VALUES 'a', 'c'")
+
+    spark.sql(
+        f"""
+            CREATE OR REPLACE TABLE {catalog_name}.default.test_incremental_read (
+                dt     date,
+                number integer,
+                letter string
+            )
+            USING iceberg
+            TBLPROPERTIES (
+                'format-version'='2'
+            );
+            """
+    )
+
+    spark.sql(
+        f"""
+            INSERT INTO {catalog_name}.default.test_incremental_read
+            VALUES (CAST('2022-03-01' AS date), 1, 'a')
+            """
+    )
+
+    spark.sql(
+        f"""
+            INSERT INTO {catalog_name}.default.test_incremental_read
+            VALUES (CAST('2022-03-01' AS date), 2, 'b')
+            """
+    )
+
+    spark.sql(
+        f"""
+            INSERT INTO {catalog_name}.default.test_incremental_read
+            VALUES (CAST('2022-03-02' AS date), 3, 'c'), (CAST('2022-03-02' AS date), 4, 'b')
+            """
+    )
+
+    spark.sql(
+        f"""
+            DELETE FROM {catalog_name}.default.test_incremental_read
+            WHERE number = 2
+            """
+    )
+
+    # https://github.com/apache/iceberg/issues/1092#issuecomment-638432848 / https://github.com/apache/iceberg/issues/3747#issuecomment-1145419407
+    # Don't do replace for Hive catalog as REPLACE TABLE requires certain Hive server configuration
+    if catalog_name != "hive":
+        # Replace to break snapshot lineage:
+        spark.sql(
+            f"""
+                REPLACE TABLE {catalog_name}.default.test_incremental_read
+                USING iceberg
+                TBLPROPERTIES ('format-version'='2')
+                AS SELECT number, letter FROM {catalog_name}.default.test_incremental_read
+                """
+        )

pyiceberg/manifest.py

@@ -866,6 +866,14 @@ def fetch_manifest_entry(self, io: FileIO, discard_deleted: bool = True) -> List
                 if not discard_deleted or entry.status != ManifestEntryStatus.DELETED
             ]
 
+    def __eq__(self, other: Any) -> bool:
+        """Return the equality of two instances of the ManifestFile class."""
+        return self.manifest_path == other.manifest_path if isinstance(other, ManifestFile) else False
+
+    def __hash__(self) -> int:
+        """Return the hash of manifest_path."""
+        return hash(self.manifest_path)
+
 
 @cached(cache=LRUCache(maxsize=128), key=lambda io, manifest_list: hashkey(manifest_list))
 def _manifests(io: FileIO, manifest_list: str) -> Tuple[ManifestFile, ...]: