[Kernel] Add specific exception subclasses for loadSnapshotAt version failures

kernel/kernel-api/src/main/java/io/delta/kernel/exceptions/VersionToLoadAfterLatestCommitException.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright (2026) 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.exceptions;
+
+import io.delta.kernel.annotation.Evolving;
+
+/**
+ * Thrown when the requested version to load is higher than the latest version available in the
+ * Delta log, i.e. it has not been materialized yet. This is typically recoverable: a consumer may
+ * retry once the requested version has been committed.
+ *
+ * @since 4.3.0
+ */
+@Evolving
+public class VersionToLoadAfterLatestCommitException extends KernelException {
+
+  public VersionToLoadAfterLatestCommitException(String message) {
+    super(message);
+  }
+}

kernel/kernel-api/src/main/java/io/delta/kernel/exceptions/VersionTruncatedException.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright (2026) 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.exceptions;
+
+import io.delta.kernel.annotation.Evolving;
+
+/**
+ * Thrown when the requested version to load is lower than the earliest version available in the
+ * Delta log because the log has been truncated (e.g. by manual deletion or the log/checkpoint
+ * retention policy). This is unrecoverable for the requested version.
+ *
+ * @since 4.3.0
+ */
+@Evolving
+public class VersionTruncatedException extends KernelException {
+
+  public VersionTruncatedException(String message) {
+    super(message);
+  }
+}

kernel/kernel-api/src/main/java/io/delta/kernel/internal/DeltaErrors.java

@@ -58,7 +58,7 @@ public static KernelException versionBeforeFirstAvailableCommit(
                 + "manual deletion or the log/checkpoint retention policy. The earliest available "
                 + "version is %s.",
             tablePath, versionToLoad, earliestVersion);
-    return new KernelException(message);
+    return new VersionTruncatedException(message);
   }
 
   public static KernelException versionToLoadAfterLatestCommit(
@@ -68,7 +68,7 @@ public static KernelException versionToLoadAfterLatestCommit(
             "%s: Cannot load table version %s as it does not exist. "
                 + "The latest available version is %s.",
             tablePath, versionToLoad, latestVersion);
-    return new KernelException(message);
+    return new VersionToLoadAfterLatestCommitException(message);
   }
 
   public static KernelException timestampBeforeFirstAvailableCommit(

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/SnapshotManagerSuite.scala

@@ -23,7 +23,7 @@ import scala.reflect.ClassTag
 
 import io.delta.kernel.data.{ColumnarBatch, ColumnVector}
 import io.delta.kernel.engine.FileReadResult
-import io.delta.kernel.exceptions.{InvalidTableException, TableNotFoundException}
+import io.delta.kernel.exceptions.{InvalidTableException, TableNotFoundException, VersionToLoadAfterLatestCommitException, VersionTruncatedException}
 import io.delta.kernel.expressions.Predicate
 import io.delta.kernel.internal.checkpoints.{CheckpointInstance, CheckpointMetaData, SidecarFile}
 import io.delta.kernel.internal.fs.Path
@@ -411,12 +411,12 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
   }
 
   test("getLogSegmentForVersion: versionToLoad higher than possible") {
-    testExpectedError[RuntimeException](
+    testExpectedError[VersionToLoadAfterLatestCommitException](
       files = deltaFileStatuses(Seq(0L)),
       versionToLoad = Optional.of(15),
       expectedErrorMessageContains =
         "Cannot load table version 15 as it does not exist. The latest available version is 0")
-    testExpectedError[RuntimeException](
+    testExpectedError[VersionToLoadAfterLatestCommitException](
       files = deltaFileStatuses((10L until 13L)) ++ singularCheckpointFileStatuses(Seq(10L)),
       versionToLoad = Optional.of(15),
       expectedErrorMessageContains =
@@ -475,7 +475,7 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
   }
 
   test("getLogSegmentForVersion: versionToLoad not constructable from history") {
-    testExpectedError[RuntimeException](
+    testExpectedError[VersionTruncatedException](
       deltaFileStatuses(20L until 25L) ++ singularCheckpointFileStatuses(Seq(20L)),
       versionToLoad = Optional.of(15),
       expectedErrorMessageContains = "Cannot load table version 15")