From 2f6cd9815731d1397f1ed06c590c80c3c01f942e Mon Sep 17 00:00:00 2001 From: shekharrajak Date: Tue, 24 Mar 2026 22:56:10 +0530 Subject: [PATCH 1/2] bench: improve iceberg tpc benchmarking --- benchmarks/tpc/create-iceberg-tables.py | 2 + benchmarks/tpc/engines/spark-iceberg.toml | 43 +++++++++++++++ benchmarks/tpc/tpcbench.py | 67 +++++++++++++++++++++++ 3 files changed, 112 insertions(+) create mode 100644 benchmarks/tpc/engines/spark-iceberg.toml diff --git a/benchmarks/tpc/create-iceberg-tables.py b/benchmarks/tpc/create-iceberg-tables.py index 219969bda7..3ba7cbd8b1 100644 --- a/benchmarks/tpc/create-iceberg-tables.py +++ b/benchmarks/tpc/create-iceberg-tables.py @@ -119,6 +119,8 @@ def main(benchmark: str, parquet_path: str, warehouse: str, catalog: str, databa for table in table_names: parquet_table_path = f"{parquet_path}/{table}.parquet" + if not os.path.exists(parquet_table_path): + parquet_table_path = f"{parquet_path}/{table}" iceberg_table = f"{catalog}.{database}.{table}" print(f"Converting {parquet_table_path} -> {iceberg_table}") diff --git a/benchmarks/tpc/engines/spark-iceberg.toml b/benchmarks/tpc/engines/spark-iceberg.toml new file mode 100644 index 0000000000..de04a3a6f0 --- /dev/null +++ b/benchmarks/tpc/engines/spark-iceberg.toml @@ -0,0 +1,43 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# Spark + Iceberg JVM scan (no Comet). +# Baseline for comparing against comet-iceberg (native Rust scan). + +[engine] +name = "spark-iceberg" + +[env] +required = ["ICEBERG_JAR", "ICEBERG_WAREHOUSE"] + +[env.defaults] +ICEBERG_CATALOG = "local" + +[spark_submit] +jars = ["$ICEBERG_JAR"] +driver_class_path = ["$ICEBERG_JAR"] + +[spark_conf] +"spark.driver.extraClassPath" = "$ICEBERG_JAR" +"spark.executor.extraClassPath" = "$ICEBERG_JAR" +"spark.sql.catalog.${ICEBERG_CATALOG}" = "org.apache.iceberg.spark.SparkCatalog" +"spark.sql.catalog.${ICEBERG_CATALOG}.type" = "hadoop" +"spark.sql.catalog.${ICEBERG_CATALOG}.warehouse" = "$ICEBERG_WAREHOUSE" +"spark.sql.defaultCatalog" = "${ICEBERG_CATALOG}" + +[tpcbench_args] +use_iceberg = true diff --git a/benchmarks/tpc/tpcbench.py b/benchmarks/tpc/tpcbench.py index a654474a6d..feacb3ccf1 100644 --- a/benchmarks/tpc/tpcbench.py +++ b/benchmarks/tpc/tpcbench.py @@ -71,6 +71,37 @@ def is_select_statement(sql): return first_keyword(sql) in ("select", "with") +def collect_execution_metrics(df): + """Extract execution metrics from the executed plan after df.collect(). + + Returns a dict with operator-level metrics including rows scanned, + bytes read, shuffle bytes, etc. + """ + metrics = {} + plan = df._jdf.queryExecution().executedPlan() + + def walk(node, depth=0): + name = node.nodeName() + node_metrics = {} + metric_map = node.metrics() + keys = list(metric_map.keys()) + for k in keys: + m = metric_map.apply(k) + node_metrics[k] = m.value() + if node_metrics: + key = f"{name}_{depth}" + metrics[key] = node_metrics + children = list(node.children()) + for child in children: + walk(child, depth + 1) + + try: + walk(plan) + except Exception: + pass # Best-effort metric extraction + return metrics + + def main( benchmark: str, data_path: str, @@ -83,6 +114,8 @@ def main( query_num: int = None, write_path: str = None, options: Dict[str, str] = None, + plan_dir: str = None, + metrics_dir: str = None, ): if options is None: options = {} @@ -190,6 +223,20 @@ def main( df = spark.sql(sql) df.explain("formatted") + # Save formatted plan to file before execution (benchmark feature) + if plan_dir is not None and is_query: + os.makedirs(plan_dir, exist_ok=True) + plan_path = os.path.join(plan_dir, f"{name}-q{query_label}.plan.txt") + try: + plan_str = df._jdf.queryExecution().explainString( + spark._jvm.org.apache.spark.sql.execution + .ExplainMode.fromString("formatted")) + with open(plan_path, "w") as pf: + pf.write(plan_str) + print(f"Plan saved to {plan_path}") + except Exception as e: + print(f"Warning: could not save plan: {e}") + if is_query and write_path is not None: if len(df.columns) > 0: output_path = f"{write_path}/q{query_label}" @@ -213,6 +260,16 @@ def main( query_result["row_count"] = row_count query_result["result_hash"] = row_hash + # Save execution metrics after collect() (benchmark feature) + if metrics_dir is not None: + os.makedirs(metrics_dir, exist_ok=True) + metrics = collect_execution_metrics(df) + metrics_path = os.path.join( + metrics_dir, f"{name}-q{query_label}.metrics.json") + with open(metrics_path, "w") as mf: + json.dump(metrics, mf, indent=2) + print(f"Metrics saved to {metrics_path}") + iter_end_time = time.time() print(f"\nIteration {iteration + 1} took {iter_end_time - iter_start_time:.2f} seconds") @@ -275,6 +332,14 @@ def main( "--name", required=True, help="Prefix for result file" ) + parser.add_argument( + "--plan-dir", + help="Optional directory to write formatted plans per query" + ) + parser.add_argument( + "--metrics-dir", + help="Optional directory to write execution metrics per query" + ) parser.add_argument( "--query", type=int, help="Specific query number (1-based). If omitted, run all." @@ -297,4 +362,6 @@ def main( args.query, args.write, args.options, + args.plan_dir, + args.metrics_dir, ) From 58379df0a959a23d19ba182dde6fab8f27ad8c67 Mon Sep 17 00:00:00 2001 From: shekharrajak Date: Tue, 24 Mar 2026 23:07:03 +0530 Subject: [PATCH 2/2] bench: drop empty metrics export --- benchmarks/tpc/tpcbench.py | 47 -------------------------------------- 1 file changed, 47 deletions(-) diff --git a/benchmarks/tpc/tpcbench.py b/benchmarks/tpc/tpcbench.py index feacb3ccf1..349e837f46 100644 --- a/benchmarks/tpc/tpcbench.py +++ b/benchmarks/tpc/tpcbench.py @@ -71,37 +71,6 @@ def is_select_statement(sql): return first_keyword(sql) in ("select", "with") -def collect_execution_metrics(df): - """Extract execution metrics from the executed plan after df.collect(). - - Returns a dict with operator-level metrics including rows scanned, - bytes read, shuffle bytes, etc. - """ - metrics = {} - plan = df._jdf.queryExecution().executedPlan() - - def walk(node, depth=0): - name = node.nodeName() - node_metrics = {} - metric_map = node.metrics() - keys = list(metric_map.keys()) - for k in keys: - m = metric_map.apply(k) - node_metrics[k] = m.value() - if node_metrics: - key = f"{name}_{depth}" - metrics[key] = node_metrics - children = list(node.children()) - for child in children: - walk(child, depth + 1) - - try: - walk(plan) - except Exception: - pass # Best-effort metric extraction - return metrics - - def main( benchmark: str, data_path: str, @@ -115,7 +84,6 @@ def main( write_path: str = None, options: Dict[str, str] = None, plan_dir: str = None, - metrics_dir: str = None, ): if options is None: options = {} @@ -260,16 +228,6 @@ def main( query_result["row_count"] = row_count query_result["result_hash"] = row_hash - # Save execution metrics after collect() (benchmark feature) - if metrics_dir is not None: - os.makedirs(metrics_dir, exist_ok=True) - metrics = collect_execution_metrics(df) - metrics_path = os.path.join( - metrics_dir, f"{name}-q{query_label}.metrics.json") - with open(metrics_path, "w") as mf: - json.dump(metrics, mf, indent=2) - print(f"Metrics saved to {metrics_path}") - iter_end_time = time.time() print(f"\nIteration {iteration + 1} took {iter_end_time - iter_start_time:.2f} seconds") @@ -336,10 +294,6 @@ def main( "--plan-dir", help="Optional directory to write formatted plans per query" ) - parser.add_argument( - "--metrics-dir", - help="Optional directory to write execution metrics per query" - ) parser.add_argument( "--query", type=int, help="Specific query number (1-based). If omitted, run all." @@ -363,5 +317,4 @@ def main( args.write, args.options, args.plan_dir, - args.metrics_dir, )