working write bencmark

Author: OussamaSaoudi

kernel/kernel-defaults/src/test/java/io/delta/kernel/defaults/benchmarks/AbstractBenchmarkState.java

@@ -40,14 +40,14 @@ public abstract class AbstractBenchmarkState {
    * dynamically by JMH. The value is set in the main method.
    */
   @Param({})
-  private String workloadSpecJson;
+  public String workloadSpecJson;
 
   /**
    * The engine to use for this benchmark. Note: This parameter will be set dynamically by JMH. The
    * value is set in the main method.
    */
   @Param({})
-  private String engineName;
+  public String engineName;
 
   /** The workload runner initialized for this benchmark invocation. */
   private WorkloadRunner runner;
@@ -76,6 +76,17 @@ public void setupInvocation() throws Exception {
     runner.setup();
   }
 
+  /**
+   * Teardown method that runs after each benchmark invocation. This calls the {@link
+   * WorkloadRunner#cleanup()} to clean up any state created during execution.
+   *
+   * @throws Exception If any error occurs during cleanup.
+   */
+  @TearDown(Level.Invocation)
+  public void teardownInvocation() throws Exception {
+    runner.cleanup();
+  }
+
   /**
    * Returns an instance of the desired engine based on the provided engine name.
    *

kernel/kernel-defaults/src/test/java/io/delta/kernel/defaults/benchmarks/BenchmarkUtils.java

@@ -115,11 +115,10 @@ private static List<WorkloadSpec> loadSpecsFromTable(Path tableDir) {
     validateTableStructure(tableDir);
 
     Path tableInfoPath = tableDir.resolve(TABLE_INFO_FILE_NAME);
-    Path deltaDir = tableDir.resolve(DELTA_DIR_NAME);
     Path specsDir = tableDir.resolve(SPECS_DIR_NAME);
 
     TableInfo tableInfo =
-        TableInfo.fromJsonPath(tableInfoPath.toString(), deltaDir.toAbsolutePath().toString());
+        TableInfo.fromJsonPath(tableInfoPath.toString(), tableDir.toAbsolutePath().toString());
 
     return findSpecDirectories(specsDir).stream()
         .map(specDir -> loadSingleSpec(specDir, tableInfo))

kernel/kernel-defaults/src/test/java/io/delta/kernel/defaults/benchmarks/WorkloadBenchmark.java

@@ -33,6 +33,8 @@
 import org.openjdk.jmh.runner.options.Options;
 import org.openjdk.jmh.runner.options.OptionsBuilder;
 import org.openjdk.jmh.runner.options.TimeValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Generic JMH benchmark for all workload types. Automatically loads and runs benchmarks based on
@@ -45,12 +47,23 @@
 @Measurement(iterations = 5, time = 1)
 public class WorkloadBenchmark<T> {
 
+  private static final Logger log = LoggerFactory.getLogger(WorkloadBenchmark.class);
+
   /** Default implementation of BenchmarkState that supports only the "default" engine. */
   public static class DefaultBenchmarkState extends AbstractBenchmarkState {
     @Override
     protected Engine getEngine(String engineName) {
       if (engineName.equals("default")) {
-        return DefaultEngine.create(new Configuration());
+        return DefaultEngine.create(
+            new Configuration() {
+              {
+                // Set the batch sizes to small so that we get to test the multiple batch/file
+                // scenarios.
+                set("delta.kernel.default.parquet.reader.batch-size", "20");
+                set("delta.kernel.default.json.reader.batch-size", "20");
+                set("delta.kernel.default.parquet.writer.targetMaxFileSize", "20");
+              }
+            });
       } else {
         throw new IllegalArgumentException("Unsupported engine: " + engineName);
       }
@@ -78,6 +91,7 @@ public void benchmarkWorkload(DefaultBenchmarkState state, Blackhole blackhole)
   public static void main(String[] args) throws RunnerException, IOException {
     // Get workload specs from the workloads directory
     List<WorkloadSpec> workloadSpecs = BenchmarkUtils.loadAllWorkloads(WORKLOAD_SPECS_DIR);
+    System.out.println("Loaded " + workloadSpecs.size() + " workload specs");
     if (workloadSpecs.isEmpty()) {
       throw new RunnerException(
           "No workloads found. Please add workload specs to the workloads directory.");
@@ -87,7 +101,12 @@ public static void main(String[] args) throws RunnerException, IOException {
     List<WorkloadSpec> filteredSpecs = new ArrayList<>();
     for (WorkloadSpec spec : workloadSpecs) {
       // TODO: In the future, we can filter specific workloads using command line args here.
-      filteredSpecs.addAll(spec.getWorkloadVariants());
+      List<WorkloadSpec> variants = spec.getWorkloadVariants();
+      for (WorkloadSpec variant : variants) {
+        if (variant.getType().equals("write")) {
+          filteredSpecs.add(variant);
+        }
+      }
     }
 
     // Convert paths into a String array for JMH. JMH requires that parameters be of type String[].
@@ -102,13 +121,13 @@ public static void main(String[] args) throws RunnerException, IOException {
             // TODO: In the future, this can be extended to support multiple engines.
             .param("engineName", "default")
             .forks(1)
-            .warmupIterations(3) // Proper warmup for production benchmarks
-            .measurementIterations(5) // Proper measurement iterations for production benchmarks
+            .warmupIterations(3)
+            .measurementIterations(5)
             .warmupTime(TimeValue.seconds(1))
             .measurementTime(TimeValue.seconds(1))
             .addProfiler(KernelMetricsProfiler.class)
             .build();
 
-    new Runner(opt, new WorkloadOutputFormat()).run();
+    new Runner(opt).run();
   }
 }

kernel/kernel-defaults/src/test/java/io/delta/kernel/defaults/benchmarks/models/TableInfo.java

@@ -16,10 +16,12 @@
 
 package io.delta.kernel.defaults.benchmarks.models;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.File;
 import java.io.IOException;
-import org.codehaus.jackson.map.ObjectMapper;
+import java.nio.file.Paths;
 
 /**
  * Represents metadata about a Delta table used in benchmark workloads.
@@ -30,17 +32,26 @@
  *
  * <p>TableInfo instances are typically loaded from JSON files in the workload specifications
  * directory structure. Each table directory should contain a {@code table_info.json} file with the
- * table metadata and a {@code delta} subdirectory containing the actual table data. The table root
- * path is the absolute path to the root of the table and is provided separately in {@link
- * WorkloadSpec#fromJsonPath(String, String, TableInfo)}.
+ * table metadata and a {@code delta} subdirectory containing the actual table data.
  *
- * <p>Example JSON structure:
+ * <p>Example JSON structure for relative table (default):
  *
  * <pre>{@code
  * {
- *   "name": "large-table",
- *   "description": "A large Delta table with multi-part checkpoints for performance testing",
- *   "engineInfo": "Apache-Spark/3.5.1 Delta-Lake/3.1.0"
+ *   "name": "basic_table",
+ *   "description": "A basic Delta table for benchmarking",
+ *   "engine_info": "Apache-Spark/3.5.1 Delta-Lake/3.1.0"
+ * }
+ * }</pre>
+ *
+ * <p>Example JSON structure for absolute path table:
+ *
+ * <pre>{@code
+ * {
+ *   "name": "s3_table",
+ *   "description": "Table stored in S3",
+ *   "table_type": "absolute",
+ *   "table_path": "s3://my-bucket/path/to/table"
  * }
  * }</pre>
  */
@@ -60,23 +71,28 @@ public class TableInfo {
   @JsonProperty("engine_info")
   public String engineInfo;
 
-  /** The root path where the Delta table is stored. */
-  @JsonProperty("table_root")
-  public String tableRoot;
+  /**
+   * The type of table location: "relative" (default) or "absolute".
+   *
+   * <p>When "relative", the table is located at {table_info_directory}/delta. When "absolute", the
+   * table is located at the path specified in {@link #tablePath}.
+   */
+  @JsonProperty("table_type")
+  private String tableType;
 
-  /** @return the absolute path to the root of the table */
-  public String getTableRoot() {
-    return tableRoot;
-  }
+  /**
+   * The absolute path to the table when {@link #tableType} is "absolute". Can be a local path
+   * (file:///) or S3 path (s3://). Null when table_type is "relative".
+   */
+  @JsonProperty("table_path")
+  private String tablePath;
 
   /**
-   * Sets the root path of the Delta table.
-   *
-   * @param tableRoot the absolute path to the root of the table
+   * The resolved absolute path to the root of the table. This is computed after deserialization
+   * based on {@link #tableType} and {@link #tablePath}.
    */
-  public void setTableRoot(String tableRoot) {
-    this.tableRoot = tableRoot;
-  }
+  @JsonProperty("table_info_path")
+  private String tableInfoPath;
 
   /**
    * Default constructor for Jackson deserialization.
@@ -86,23 +102,47 @@ public void setTableRoot(String tableRoot) {
    */
   public TableInfo() {}
 
+  /** Resolves the table root path based on the table type and location configuration. */
+  @JsonIgnore
+  public String getResolvedTableRoot() {
+    if ("absolute".equals(tableType)) {
+      if (tablePath == null || tablePath.trim().isEmpty()) {
+        throw new IllegalStateException(
+            "table_path must be specified when table_type is 'absolute'");
+      }
+      return tablePath;
+    } else {
+      // Default to "relative" if tableType is null or "relative"
+      return Paths.get(tableInfoPath, "delta").toAbsolutePath().toString();
+    }
+  }
+
+  public String getTableInfoPath() {
+    return tableInfoPath;
+  }
+
+  public void setTableInfoPath(String tableInfoDirectory) {
+    this.tableInfoPath = tableInfoDirectory;
+  }
+
   /**
    * Creates a TableInfo instance by reading from a JSON file at the specified path.
    *
-   * <p>This method loads table metadata from a JSON file and sets the table root path. The JSON
-   * file should contain the table name and description, while the table root path is provided
-   * separately with the absolute path.
+   * <p>This method loads table metadata from a JSON file and resolves the table root path. The JSON
+   * file should contain the table name and description, while the table root path is computed based
+   * on the table_type and table_path fields.
    *
    * @param jsonPath the path to the JSON file containing the TableInfo metadata
-   * @param tableRoot the absolute path to the root of the Delta table
-   * @return a TableInfo instance populated from the JSON file and table root path
+   * @param tableInfoPath the directory containing the table_info.json file (used for relative path
+   *     resolution)
+   * @return a TableInfo instance populated from the JSON file with resolved table root path
    * @throws RuntimeException if there is an error reading or parsing the JSON file
    */
-  public static TableInfo fromJsonPath(String jsonPath, String tableRoot) {
+  public static TableInfo fromJsonPath(String jsonPath, String tableInfoPath) {
     ObjectMapper mapper = new ObjectMapper();
     try {
       TableInfo info = mapper.readValue(new File(jsonPath), TableInfo.class);
-      info.setTableRoot(tableRoot);
+      info.setTableInfoPath(tableInfoPath);
       return info;
     } catch (IOException e) {
       throw new RuntimeException("Failed to read TableInfo from JSON file: " + jsonPath, e);
@@ -112,8 +152,8 @@ public static TableInfo fromJsonPath(String jsonPath, String tableRoot) {
   /**
    * Returns a string representation of this TableInfo.
    *
-   * <p>The string includes the table name, description, and engine info, but excludes the table
-   * root path for security reasons (as it may contain sensitive path information).
+   * <p>The string includes the table name, description, engine info, and CCv2 status, but excludes
+   * the table root path for security reasons (as it may contain sensitive path information).
    *
    * @return a string representation of this TableInfo
    */
@@ -125,6 +165,8 @@ public String toString() {
         + description
         + "', engineInfo='"
         + engineInfo
-        + "'}";
+        + "', tableType='"
+        + tableType
+        + "}";
   }
 }

kernel/kernel-defaults/src/test/java/io/delta/kernel/defaults/benchmarks/models/WorkloadSpec.java

@@ -33,7 +33,10 @@
  * field in the JSON.
  */
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
-@JsonSubTypes({@JsonSubTypes.Type(value = ReadSpec.class, name = "read")})
+@JsonSubTypes({
+  @JsonSubTypes.Type(value = ReadSpec.class, name = "read"),
+  @JsonSubTypes.Type(value = WriteSpec.class, name = "write")
+})
 public abstract class WorkloadSpec {
   /**
    * The type of workload (e.g., "read"). This is used by Jackson's polymorphic deserialization to