bench: improve Iceberg TPC workflow and plan capture

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}")

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

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,
     )