[Kernel][Kernel UC] Make startBoundary required for CommitRange (#5537)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [ ] Spark
- [ ] Standalone
- [ ] Flink
- [X] Kernel
- [ ] Other (fill in here)

## Description

Currently when building a commit range start boundary is optional and
defaults to 0 when not provided. However, there is no good use-case
during which the start version is not provided so this PR makes it
required. The prior behavior can always be implemented by the caller by
providing version=0 start boundary.

## How was this patch tested?

Updates/adds tests.

Author: allisonport-db

kernel-spark/src/main/java/io/delta/kernel/spark/snapshot/PathBasedSnapshotManager.java

@@ -131,8 +131,8 @@ public void checkVersionExists(long version, boolean mustBeRecreatable, boolean
   @Override
   public CommitRange getTableChanges(Engine engine, long startVersion, Optional<Long> endVersion) {
     CommitRangeBuilder builder =
-        TableManager.loadCommitRange(tablePath)
-            .withStartBoundary(CommitRangeBuilder.CommitBoundary.atVersion(startVersion));
+        TableManager.loadCommitRange(
+            tablePath, CommitRangeBuilder.CommitBoundary.atVersion(startVersion));
 
     if (endVersion.isPresent()) {
       builder =

kernel/kernel-api/src/main/java/io/delta/kernel/CommitRange.java

@@ -59,10 +59,9 @@ public interface CommitRange {
    * <p>The boundary indicates whether the range was defined using a specific version number or a
    * timestamp.
    *
-   * @return an {@link Optional} containing the start boundary, or empty if the range was created
-   *     with default start parameters (version 0)
+   * @return the start boundary for this commit range
    */
-  Optional<CommitRangeBuilder.CommitBoundary> getQueryStartBoundary();
+  CommitRangeBuilder.CommitBoundary getQueryStartBoundary();
 
   /**
    * Returns the original query boundary used to define the end boundary of this commit range, if

kernel/kernel-api/src/main/java/io/delta/kernel/CommitRangeBuilder.java

@@ -29,25 +29,15 @@
  * A builder for creating {@link CommitRange} instances that define a contiguous range of commits in
  * a Delta Lake table.
  *
- * <p>If no start specification is provided, the range defaults to starting at version 0. If no end
- * specification is provided, the range defaults to the latest available version.
+ * <p>The start boundary is required and provided via {@link TableManager#loadCommitRange(String,
+ * CommitBoundary)}. If no end specification is provided, the range defaults to the latest available
+ * version.
  *
  * @since 3.4.0
  */
 @Experimental
 public interface CommitRangeBuilder {
 
-  /**
-   * Configures the builder to start the commit range at a specific version or timestamp.
-   *
-   * <p>If not specified, the commit range will default to starting at version 0.
-   *
-   * @param startBoundary the boundary specification for the start of the commit range, must not be
-   *     null
-   * @return this builder instance configured with the specified start boundary
-   */
-  CommitRangeBuilder withStartBoundary(CommitBoundary startBoundary);
-
   /**
    * Configures the builder to end the commit range at a specific version or timestamp.
    *

kernel/kernel-api/src/main/java/io/delta/kernel/TableManager.java

@@ -64,14 +64,17 @@ static CreateTableTransactionBuilder buildCreateTableTransaction(
   /**
    * Creates a builder for loading a CommitRange at a given path.
    *
-   * <p>The returned builder can be configured with start version or timestamp and an end version or
-   * timestamp, and with additional metadata to optimize the loading process.
+   * <p>The returned builder can be configured with an end version or timestamp, and with additional
+   * metadata to optimize the loading process.
    *
    * @param path the file system path to the Delta table
+   * @param startBoundary the boundary specification for the start of the commit range, must not be
+   *     null
    * @return a {@link CommitRangeBuilder} that can be used to load a {@link CommitRange} at the
    *     given path
    */
-  static CommitRangeBuilder loadCommitRange(String path) {
-    return new CommitRangeBuilderImpl(path);
+  static CommitRangeBuilder loadCommitRange(
+      String path, CommitRangeBuilder.CommitBoundary startBoundary) {
+    return new CommitRangeBuilderImpl(path, startBoundary);
   }
 }

kernel/kernel-api/src/main/java/io/delta/kernel/internal/commitrange/CommitRangeBuilderImpl.java

@@ -38,31 +38,26 @@ public class CommitRangeBuilderImpl implements CommitRangeBuilder {
 
   public static class Context {
     public final String unresolvedPath;
-    public Optional<CommitBoundary> startBoundaryOpt = Optional.empty();
+    public final CommitBoundary startBoundary;
     public Optional<CommitBoundary> endBoundaryOpt = Optional.empty();
     public List<ParsedLogData> logDatas = Collections.emptyList();
 
-    public Context(String unresolvedPath) {
+    public Context(String unresolvedPath, CommitBoundary startBoundary) {
       this.unresolvedPath = requireNonNull(unresolvedPath, "unresolvedPath is null");
+      this.startBoundary = requireNonNull(startBoundary, "startBoundary is null");
     }
   }
 
   private final Context ctx;
 
-  public CommitRangeBuilderImpl(String unresolvedPath) {
-    ctx = new Context(unresolvedPath);
+  public CommitRangeBuilderImpl(String unresolvedPath, CommitBoundary startBoundary) {
+    ctx = new Context(unresolvedPath, startBoundary);
   }
 
   ///////////////////////////////////////
   // Public CommitRangeBuilder Methods //
   ///////////////////////////////////////
 
-  @Override
-  public CommitRangeBuilderImpl withStartBoundary(CommitBoundary startBoundary) {
-    ctx.startBoundaryOpt = Optional.of(requireNonNull(startBoundary, "startBoundary is null"));
-    return this;
-  }
-
   @Override
   public CommitRangeBuilderImpl withEndBoundary(CommitBoundary endBoundary) {
     ctx.endBoundaryOpt = Optional.of(requireNonNull(endBoundary, "endBoundary is null"));
@@ -86,9 +81,10 @@ public CommitRange build(Engine engine) {
   ////////////////////////////
 
   private void validateInputOnBuild() {
-    // Validate that start boundary is less than or equal to end boundary if both are provided
-    if (ctx.startBoundaryOpt.isPresent() && ctx.endBoundaryOpt.isPresent()) {
-      CommitBoundary startBoundary = ctx.startBoundaryOpt.get();
+    // Validate that start boundary is less than or equal to end boundary if end boundary is
+    // provided
+    if (ctx.endBoundaryOpt.isPresent()) {
+      CommitBoundary startBoundary = ctx.startBoundary;
       CommitBoundary endBoundary = ctx.endBoundaryOpt.get();
 
       // If both are version-based, compare versions

kernel/kernel-api/src/main/java/io/delta/kernel/internal/commitrange/CommitRangeFactory.java

@@ -69,28 +69,22 @@ CommitRangeImpl create(Engine engine) {
       logger.info("{}: Resolved end-boundary to the latest version {}", tablePath, endVersion);
     }
     return new CommitRangeImpl(
-        tablePath, ctx.startBoundaryOpt, ctx.endBoundaryOpt, startVersion, endVersion, deltas);
+        tablePath, ctx.startBoundary, ctx.endBoundaryOpt, startVersion, endVersion, deltas);
   }
 
   private long resolveStartVersion(Engine engine, List<ParsedCatalogCommitData> catalogCommits) {
-    if (!ctx.startBoundaryOpt.isPresent()) {
-      // Default to version 0 if no start boundary is provided
-      return 0L;
-    }
-    CommitRangeBuilder.CommitBoundary startBoundary = ctx.startBoundaryOpt.get();
-
-    if (startBoundary.isVersion()) {
-      return startBoundary.getVersion();
+    if (ctx.startBoundary.isVersion()) {
+      return ctx.startBoundary.getVersion();
     } else {
       logger.info(
           "{}: Trying to resolve start-boundary timestamp {} to version",
           tablePath,
-          startBoundary.getTimestamp());
+          ctx.startBoundary.getTimestamp());
       return DeltaHistoryManager.getVersionAtOrAfterTimestamp(
           engine,
           logPath,
-          startBoundary.getTimestamp(),
-          (SnapshotImpl) startBoundary.getLatestSnapshot(),
+          ctx.startBoundary.getTimestamp(),
+          (SnapshotImpl) ctx.startBoundary.getLatestSnapshot(),
           catalogCommits);
     }
   }
@@ -143,7 +137,7 @@ private void logResolvedVersions(long startVersion, Optional<Long> endVersionOpt
         tablePath,
         startVersion,
         endVersionOpt,
-        ctx.startBoundaryOpt,
+        ctx.startBoundary,
         ctx.endBoundaryOpt);
   }
 

kernel/kernel-api/src/main/java/io/delta/kernel/internal/commitrange/CommitRangeImpl.java

@@ -41,7 +41,7 @@
 public class CommitRangeImpl implements CommitRange {
 
   private final Path dataPath;
-  private final Optional<CommitRangeBuilder.CommitBoundary> startBoundaryOpt;
+  private final CommitRangeBuilder.CommitBoundary startBoundary;
   private final Optional<CommitRangeBuilder.CommitBoundary> endBoundaryOpt;
 
   private final long startVersion;
@@ -50,7 +50,7 @@ public class CommitRangeImpl implements CommitRange {
 
   public CommitRangeImpl(
       Path dataPath,
-      Optional<CommitRangeBuilder.CommitBoundary> startBoundaryOpt,
+      CommitRangeBuilder.CommitBoundary startBoundary,
       Optional<CommitRangeBuilder.CommitBoundary> endBoundaryOpt,
       long startVersion,
       long endVersion,
@@ -59,7 +59,7 @@ public CommitRangeImpl(
     checkArgument(
         deltas.size() == endVersion - startVersion + 1, "deltaFiles size must match size of range");
     this.dataPath = requireNonNull(dataPath, "dataPath cannot be null");
-    this.startBoundaryOpt = requireNonNull(startBoundaryOpt, "startSpecOpt cannot be null");
+    this.startBoundary = requireNonNull(startBoundary, "startBoundary cannot be null");
     this.endBoundaryOpt = requireNonNull(endBoundaryOpt, "endSpecOpt cannot be null");
     this.startVersion = startVersion;
     this.endVersion = endVersion;
@@ -81,8 +81,8 @@ public long getEndVersion() {
   }
 
   @Override
-  public Optional<CommitRangeBuilder.CommitBoundary> getQueryStartBoundary() {
-    return startBoundaryOpt;
+  public CommitRangeBuilder.CommitBoundary getQueryStartBoundary() {
+    return startBoundary;
   }
 
   @Override