Use consistent variables in TPCH

benchmarks/tpch/bodo/dataframe_queries.py

@@ -956,7 +956,12 @@ def run_queries(root: str, queries: list[int], scale_factor: float, backend):
 
     total_start = time.time()
     for query in queries:
-        globals()[f"q{query:02}"](*queries_to_args[query])
+        query_func = globals()[f"q{query:02}"]
+        # Warm up run:
+        query_func(*queries_to_args[query])
+
+        # Second run for timing:
+        query_func(*queries_to_args[query])
     print(f"Total query execution time (s): {time.time() - total_start}")
 
 

benchmarks/tpch/bodo/run_bodo.py

@@ -8,7 +8,7 @@
 
 https://docs.bodo.ai/latest/guides/using_bodo_platform/bodo_platform_sdk_guide/#installation
 
-NOTE: This script assumes that you have the file `dataframe_lib.py`
+NOTE: This script assumes that you have the file `dataframe_queries.py`
 copied in your current workspace.
 
 usage:
@@ -19,8 +19,8 @@
 
 from bodosdk import BodoWorkspaceClient
 
-NUM_WORKERS = 2
-WORKER_INSTANCE = "r6i.8xlarge"
+NUM_WORKERS = 4
+WORKER_INSTANCE = "r6i.16xlarge"
 
 
 def run_bodo_benchmark(folder, queries, scale_factor):
@@ -32,22 +32,21 @@ def run_bodo_benchmark(folder, queries, scale_factor):
     )
     benchmark_cluster.wait_for_status(["RUNNING"])
 
-    args = {
-        "folder": folder,
-        "scale_factor": str(scale_factor),
-    }
-    if queries:
-        args["queries"] = " ".join(str(q) for q in queries)
-
-    arg_str = " ".join(f"--{key} {value}" for key, value in args.items())
-
-    benchmark_job = benchmark_cluster.run_job(
-        code_type="PYTHON",
-        source={"type": "WORKSPACE", "path": "/"},
-        exec_file="dataframe_queries.py",
-        args=arg_str,
-    )
-    print(benchmark_job.wait_for_status(["SUCCEEDED"]).get_stdout())
+    for query in queries:
+        args = {
+            "folder": folder,
+            "scale_factor": str(scale_factor),
+            "queries": str(query),
+        }
+        arg_str = " ".join(f"--{key} {value}" for key, value in args.items())
+
+        benchmark_job = benchmark_cluster.run_job(
+            code_type="PYTHON",
+            source={"type": "WORKSPACE", "path": "/"},
+            exec_file="dataframe_queries.py",
+            args=arg_str,
+        )
+        print(benchmark_job.wait_for_status(["SUCCEEDED"]).get_stdout())
 
     # cleanup:
     benchmark_cluster.stop(wait=True)
@@ -77,7 +76,10 @@ def main():
         help="Scale factor (used in query 11).",
     )
     args = parser.parse_args()
-    run_bodo_benchmark(args.folder, args.queries, args.scale_factor)
+    queries = list(range(1, 23))
+    if args.queries is not None:
+        queries = args.queries
+    run_bodo_benchmark(args.folder, queries, args.scale_factor)
 
 
 if __name__ == "__main__":

benchmarks/tpch/dask/dask_queries.py

@@ -17,7 +17,7 @@
     "security": False,
     "n_workers": 4,
     "scheduler_instance_type": "c6i.xlarge",
-    "worker_instance_type": "c6i.16xlarge",
+    "worker_instance_type": "r6i.16xlarge",
     "docker_image": "daskdev/dask:latest",
     # Profile with AmazonS3FullAccess
     "iam_instance_profile": {"Name": "dask-benchmark"},

benchmarks/tpch/pds-benchmark/queries/common_utils.py

@@ -24,6 +24,8 @@ def get_table_path(table_name: str) -> str:
     ext = settings.run.io_type if settings.run.include_io else "parquet"
     # Bodo Change: Parquet path
     multi_part_tables = {"lineitem", "orders", "customer", "part", "partsupp"}
+    if settings.scale_factor >= 1000:
+        multi_part_tables.add("supplier")
     glob_str = "/*" if ext == "parquet" and table_name in multi_part_tables else ""
     ext = "pq" if ext == "parquet" else ext
     return f"{settings.dataset_base_dir}/{table_name}.{ext}{glob_str}"

benchmarks/tpch/pds-benchmark/queries/polars/q10.py

@@ -25,8 +25,8 @@ def q(
     assert nation is not None
     assert orders is not None
 
-    var1 = date(1993, 10, 1)
-    var2 = date(1994, 1, 1)
+    var1 = date(1994, 11, 1)
+    var2 = date(1995, 2, 1)
 
     return (
         customer.join(orders, left_on="c_custkey", right_on="o_custkey")

benchmarks/tpch/pds-benchmark/queries/polars/q14.py

@@ -19,8 +19,8 @@ def q(
     assert lineitem is not None
     assert part is not None
 
-    var1 = date(1995, 9, 1)
-    var2 = date(1995, 10, 1)
+    var1 = date(1994, 3, 1)
+    var2 = date(1994, 4, 1)
 
     return (
         lineitem.join(part, left_on="l_partkey", right_on="p_partkey")

benchmarks/tpch/pds-benchmark/queries/polars/q20.py

@@ -28,10 +28,10 @@ def q(
     assert partsupp is not None
     assert supplier is not None
 
-    var1 = date(1994, 1, 1)
-    var2 = date(1995, 1, 1)
-    var3 = "CANADA"
-    var4 = "forest"
+    var1 = date(1996, 1, 1)
+    var2 = date(1997, 1, 1)
+    var3 = "JORDAN"
+    var4 = "azure"
 
     q1 = (
         lineitem.filter(pl.col("l_shipdate").is_between(var1, var2, closed="left"))

benchmarks/tpch/pds-benchmark/queries/polars/q4.py

@@ -19,8 +19,8 @@ def q(
     assert lineitem is not None
     assert orders is not None
 
-    var1 = date(1993, 7, 1)
-    var2 = date(1993, 10, 1)
+    var1 = date(1993, 8, 1)
+    var2 = date(1993, 11, 1)
 
     return (
         # SQL exists translates to semi join in Polars API

benchmarks/tpch/pds-benchmark/queries/polars/q5.py

@@ -34,8 +34,8 @@ def q(
     assert orders is not None
 
     var1 = "ASIA"
-    var2 = date(1994, 1, 1)
-    var3 = date(1995, 1, 1)
+    var2 = date(1996, 1, 1)
+    var3 = date(1997, 1, 1)
 
     return (
         region.join(nation, left_on="r_regionkey", right_on="n_regionkey")

benchmarks/tpch/pds-benchmark/queries/polars/q6.py

@@ -14,10 +14,10 @@ def q(
     if lineitem is None:
         lineitem = utils.get_line_item_ds()
 
-    var1 = date(1994, 1, 1)
-    var2 = date(1995, 1, 1)
-    var3 = 0.05
-    var4 = 0.07
+    var1 = date(1996, 1, 1)
+    var2 = date(1997, 1, 1)
+    var3 = 0.08
+    var4 = 0.1
     var5 = 24
 
     return (

benchmarks/tpch/pds-benchmark/queries/polars/q9.py

@@ -41,7 +41,7 @@ def q(
         )
         .join(orders, left_on="l_orderkey", right_on="o_orderkey")
         .join(nation, left_on="s_nationkey", right_on="n_nationkey")
-        .filter(pl.col("p_name").str.contains("green"))
+        .filter(pl.col("p_name").str.contains("ghost"))
         .select(
             pl.col("n_name").alias("nation"),
             pl.col("o_orderdate").dt.year().alias("o_year"),

benchmarks/tpch/pyspark/emr.tf

@@ -61,11 +61,11 @@ resource "aws_emr_cluster" "emr_cluster" {
   }
 
   master_instance_group {
-    instance_type = "c6i.16xlarge"
+    instance_type = "r6i.16xlarge"
   }
 
   core_instance_group {
-    instance_type  = "c6i.16xlarge"
+    instance_type  = "r6i.16xlarge"
     instance_count = 3
 
     ebs_config {