test: update perf test table config and benches to be more useful whe…

…n measuring performance of row group filtering and row selection

crates/iceberg/benches/table_scan_execute_query.rs

@@ -43,7 +43,7 @@ pub fn bench_read_all_files_some_rows(c: &mut Criterion) {
     let table = setup(&runtime);
     let scan = table
         .scan()
-        .with_filter(Reference::new("passenger_count").greater_than(Datum::double(1.0)))
+        .with_filter(Reference::new("fare_amount").greater_than(Datum::double(50.0)))
         .build()
         .unwrap();
     let tasks = create_file_plan(&runtime, scan);
@@ -98,7 +98,7 @@ pub fn bench_read_some_files_some_rows(c: &mut Criterion) {
                     .and(Reference::new("tpep_pickup_datetime").less_than(
                         Datum::timestamptz_from_str("2024-02-02T00:00:00.000 UTC").unwrap(),
                     ))
-                    .and(Reference::new("passenger_count").greater_than(Datum::double(1.0))),
+                    .and(Reference::new("fare_amount").greater_than(Datum::double(50.0))),
             )
             .build()
             .unwrap();

crates/iceberg/benches/table_scan_plan_files.rs

@@ -57,7 +57,7 @@ async fn one_file_all_rows(table: &Table) {
 async fn all_files_some_rows(table: &Table) {
     let scan = table
         .scan()
-        .with_filter(Reference::new("passenger_count").equal_to(Datum::double(1.0)))
+        .with_filter(Reference::new("fare_amount").equal_to(Datum::double(50.0)))
         .build()
         .unwrap();
     let mut stream = scan.plan_files().await.unwrap();
@@ -79,7 +79,7 @@ async fn one_file_some_rows(table: &Table) {
                     .and(Reference::new("tpep_pickup_datetime").less_than(
                         Datum::timestamptz_from_str("2024-02-02T00:00:00.000 UTC").unwrap(),
                     ))
-                    .and(Reference::new("passenger_count").equal_to(Datum::double(1.0))),
+                    .and(Reference::new("fare_amount").equal_to(Datum::double(50.0))),
             )
             .build()
             .unwrap();

crates/iceberg/testdata/performance/spark_scripts/setup.sql

@@ -42,7 +42,12 @@ CREATE TABLE IF NOT EXISTS nyc.taxis (
 USING iceberg
 PARTITIONED BY (days(tpep_pickup_datetime));
 
-ALTER TABLE nyc.taxis WRITE ORDERED BY tpep_pickup_datetime, passenger_count;
+
+ALTER TABLE nyc.taxis SET TBLPROPERTIES (
+    'write.parquet.row-group-size-bytes'='131072',
+    'write.parquet.page-row-limit'='200'
+);
+ALTER TABLE nyc.taxis WRITE DISTRIBUTED BY PARTITION LOCALLY ORDERED BY fare_amount;
 
 CREATE TEMPORARY VIEW parquetTable
 USING org.apache.spark.sql.parquet