[Kernel] Add snapshot construction benchmarking specification

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

@@ -0,0 +1,58 @@
+/*
+ * Copyright (2025) The Delta Lake Project Authors.
+ *
+ * Licensed 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.
+ */
+
+package io.delta.kernel.defaults.benchmarks.models;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.delta.kernel.defaults.benchmarks.workloadrunners.SnapshotConstructionRunner;
+import io.delta.kernel.defaults.benchmarks.workloadrunners.WorkloadRunner;
+import io.delta.kernel.engine.Engine;
+
+public class SnapshotConstructionSpec extends WorkloadSpec {
+
+  /** The snapshot version to read. If null, the latest version will be read. From spec file. */
+  @JsonProperty("version")
+  private Long version;
+
+  /** Expected data file for validating the read data result. From spec file. */
+  @JsonProperty("expected_protocol_and_metadata")
+  private String expectedProtocolAndMetadata;
+
+  // Default constructor for Jackson
+  public SnapshotConstructionSpec() {
+    super("snapshot_construction");
+  }
+
+  // Copy constructor
+  public SnapshotConstructionSpec(
+      TableInfo tableInfo, String caseName, Long version, String expectedProtocolAndMetadata) {
+    super("snapshot_construction");
+    this.tableInfo = tableInfo;
+    this.version = version;
+    this.caseName = caseName;
+    this.expectedProtocolAndMetadata = expectedProtocolAndMetadata;
+  }
+
+  /** @return the snapshot version to read, or null if the latest version should be read. */
+  public Long getVersion() {
+    return version;
+  }
+
+  @Override
+  public WorkloadRunner getRunner(Engine engine) {
+    return new SnapshotConstructionRunner(this, engine);
+  }
+}

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 = SnapshotConstructionSpec.class, name = "snapshot_construction"),
+})
 public abstract class WorkloadSpec {
   /**
    * The type of workload (e.g., "read"). This is used by Jackson's polymorphic deserialization to

kernel/kernel-defaults/src/test/java/io/delta/kernel/defaults/benchmarks/workloadrunners/SnapshotConstructionRunner.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright (2025) The Delta Lake Project Authors.
+ *
+ * Licensed 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.
+ */
+
+package io.delta.kernel.defaults.benchmarks.workloadrunners;
+
+import io.delta.kernel.Snapshot;
+import io.delta.kernel.SnapshotBuilder;
+import io.delta.kernel.TableManager;
+import io.delta.kernel.defaults.benchmarks.models.SnapshotConstructionSpec;
+import io.delta.kernel.defaults.benchmarks.models.WorkloadSpec;
+import io.delta.kernel.engine.Engine;
+import org.openjdk.jmh.infra.Blackhole;
+
+public class SnapshotConstructionRunner extends WorkloadRunner {
+
+  private final SnapshotConstructionSpec workloadSpec;
+  private final Engine engine;
+
+  public SnapshotConstructionRunner(SnapshotConstructionSpec spec, Engine engine) {
+    this.workloadSpec = spec;
+    this.engine = engine;
+  }
+
+  @Override
+  public String getName() {
+    return "snapshot_construction";
+  }
+
+  @Override
+  public WorkloadSpec getWorkloadSpec() {
+    return this.workloadSpec;
+  }
+
+  @Override
+  public void setup() throws Exception {
+    /* No setup needed for snapshot construction */
+  }
+
+  @Override
+  public void executeAsBenchmark(Blackhole blackhole) throws Exception {
+    blackhole.consume(this.execute());
+  }
+
+  private Snapshot execute() {
+    String workloadTableRoot = workloadSpec.getTableInfo().getTableRoot();
+    SnapshotBuilder builder = TableManager.loadSnapshot(workloadTableRoot);
+    if (workloadSpec.getVersion() != null) {
+      builder.atVersion(workloadSpec.getVersion());
+    }
+    return builder.build(engine);
+  }
+}

kernel/kernel-defaults/src/test/resources/workload_specs/basic_append/specs/snapshot_latest/spec.json

@@ -0,0 +1,3 @@
+{
+  "type": "snapshot_construction"
+}