Limit parallism to reduce pressure on the memory

Author: Fokko

tests/conftest.py

@@ -2433,6 +2433,8 @@ def spark() -> "SparkSession":
     spark = (
         SparkSession.builder.appName("PyIceberg integration test")
         .config("spark.sql.session.timeZone", "UTC")
+        .config("spark.sql.shuffle.partitions", "1")
+        .config("spark.default.parallelism", "1")
         .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
         .config("spark.sql.catalog.integration", "org.apache.iceberg.spark.SparkCatalog")
         .config("spark.sql.catalog.integration.catalog-impl", "org.apache.iceberg.rest.RESTCatalog")

tests/integration/test_deletes.py

@@ -309,58 +309,57 @@ def test_delete_partitioned_table_positional_deletes_empty_batch(spark: SparkSes
     assert len(reader.read_all()) == 0
 
 
-# @pytest.mark.integration
-# @pytest.mark.filterwarnings("ignore:Merge on read is not yet supported, falling back to copy-on-write")
-# def test_read_multiple_batches_in_task_with_position_deletes(spark: SparkSession, session_catalog: RestCatalog) -> None:
-#     identifier = "default.test_read_multiple_batches_in_task_with_position_deletes"
-#     multiplier = 10
-#
-#     run_spark_commands(
-#         spark,
-#         [
-#             f"DROP TABLE IF EXISTS {identifier}",
-#             f"""
-#             CREATE TABLE {identifier} (
-#                 number              int
-#             )
-#             USING iceberg
-#             TBLPROPERTIES(
-#                 'format-version' = 2,
-#                 'write.delete.mode'='merge-on-read',
-#                 'write.update.mode'='merge-on-read',
-#                 'write.merge.mode'='merge-on-read'
-#             )
-#         """,
-#         ],
-#     )
-#
-#     tbl = session_catalog.load_table(identifier)
-#
-#     arrow_table = pa.Table.from_arrays(
-#         [
-#             pa.array(list(range(1, 1001)) * multiplier),
-#         ],
-#         schema=pa.schema([pa.field("number", pa.int32())]),
-#     )
-#
-#     tbl.append(arrow_table)
-#
-#     run_spark_commands(
-#         spark,
-#         [
-#             f"""
-#             DELETE FROM {identifier} WHERE number in (1, 2, 3, 4)
-#         """,
-#         ],
-#     )
-#
-#     tbl.refresh()
-#
-#     reader = tbl.scan(row_filter="number <= 50").to_arrow_batch_reader()
-#     assert isinstance(reader, pa.RecordBatchReader)
-#     pyiceberg_count = len(reader.read_all())
-#     expected_count = 46 * multiplier
-#     assert pyiceberg_count == expected_count, f"Failing check. {pyiceberg_count} != {expected_count}"
+@pytest.mark.integration
+@pytest.mark.filterwarnings("ignore:Merge on read is not yet supported, falling back to copy-on-write")
+def test_read_multiple_batches_in_task_with_position_deletes(spark: SparkSession, session_catalog: RestCatalog) -> None:
+    identifier = "default.test_read_multiple_batches_in_task_with_position_deletes"
+
+    run_spark_commands(
+        spark,
+        [
+            f"DROP TABLE IF EXISTS {identifier}",
+            f"""
+            CREATE TABLE {identifier} (
+                number              int
+            )
+            USING iceberg
+            TBLPROPERTIES(
+                'format-version' = 2,
+                'write.delete.mode'='merge-on-read',
+                'write.update.mode'='merge-on-read',
+                'write.merge.mode'='merge-on-read'
+            )
+        """,
+        ],
+    )
+
+    tbl = session_catalog.load_table(identifier)
+
+    arrow_table = pa.Table.from_arrays(
+        [
+            pa.array(list(range(1, 1001)) * 100),
+        ],
+        schema=pa.schema([pa.field("number", pa.int32())]),
+    )
+
+    tbl.append(arrow_table)
+
+    run_spark_commands(
+        spark,
+        [
+            f"""
+            DELETE FROM {identifier} WHERE number in (1, 2, 3, 4)
+        """,
+        ],
+    )
+
+    tbl.refresh()
+
+    reader = tbl.scan(row_filter="number <= 50").to_arrow_batch_reader()
+    assert isinstance(reader, pa.RecordBatchReader)
+    pyiceberg_count = len(reader.read_all())
+    expected_count = 46 * 100
+    assert pyiceberg_count == expected_count, f"Failing check. {pyiceberg_count} != {expected_count}"
 
 
 @pytest.mark.integration