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..349e837f46 100644 --- a/benchmarks/tpc/tpcbench.py +++ b/benchmarks/tpc/tpcbench.py @@ -83,6 +83,7 @@ def main( query_num: int = None, write_path: str = None, options: Dict[str, str] = None, + plan_dir: str = None, ): if options is None: options = {} @@ -190,6 +191,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}" @@ -275,6 +290,10 @@ 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( "--query", type=int, help="Specific query number (1-based). If omitted, run all." @@ -297,4 +316,5 @@ def main( args.query, args.write, args.options, + args.plan_dir, )