A table with positional deletes shoulds fail

crates/integration_tests/testdata/spark/entrypoint.sh

@@ -18,6 +18,8 @@
 # under the License.
 #
 
+set -e
+
 start-master.sh -p 7077
 start-worker.sh spark://spark-iceberg:7077
 start-history-server.sh

crates/integration_tests/testdata/spark/provision.py

@@ -18,7 +18,13 @@
 from pyspark.sql import SparkSession
 from pyspark.sql.functions import current_date, date_add, expr
 
-spark = SparkSession.builder.getOrCreate()
+spark = (
+    SparkSession
+        .builder
+        .config("spark.sql.shuffle.partitions", "1")
+        .config("spark.default.parallelism", "1")
+        .getOrCreate()
+)
 
 spark.sql(
     f"""

crates/integration_tests/tests/read_positional_deletes.rs

@@ -17,6 +17,7 @@
 
 //! Integration tests for rest catalog.
 
+use futures::TryStreamExt;
 use iceberg::{Catalog, TableIdent};
 use iceberg_integration_tests::set_test_fixture;
 
@@ -34,6 +35,15 @@ async fn test_read_table_with_positional_deletes() {
         .await
         .unwrap();
 
-    // 😱 If we don't support positional deletes, we should not be able to plan them
-    println!("{:?}", table.scan().build().unwrap());
+    let scan = table.scan().build().unwrap();
+    println!("{:?}", scan);
+
+    let batch_stream = scan.to_arrow().await.unwrap();
+    let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
+
+    let num_rows: usize = batches.iter().map(|v| v.num_rows()).sum();
+
+    // 😱 If we don't support positional deletes, we should fail when we try to read a table that
+    // has positional deletes. The table has 12 rows, and 2 are deleted, see provision.py
+    assert_eq!(num_rows, 10);
 }