[Kernel] Throw InvalidTableException when we encounter a known invalid table state (#3288)

#### 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

During log reconstruction we throw a bunch of various errors when we
encounter a table state that we consider invalid. This PR updates some
of those errors to be `InvalidTableException`. This doesn't necessarily
cover all possible exceptions due to an invalid table, just some of the
known and clear cases during log reconstruction.

## How was this patch tested?

Updates existing tests.

## Does this PR introduce _any_ user-facing changes?

Yes, instead of internal `RuntimeExceptions` or
`IllegalStateExceptions`, etc a `InvalidTableException` will be thrown
when certain invalid table states are encountered.

---------

Co-authored-by: Venki Korukanti <[email protected]>

Author: allisonport-db

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

@@ -0,0 +1,29 @@
+/*
+ * Copyright (2024) 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;
+
+/**
+ * Thrown when an invalid table is encountered; the table's log and/or checkpoint files are
+ * in an invalid state.
+ */
+public class InvalidTableException extends KernelException {
+
+    private static final String message = "Invalid table found at %s: %s";
+
+    public InvalidTableException(String tablePath, String reason) {
+        super(String.format(message, tablePath, reason));
+    }
+}

kernel/kernel-api/src/main/java/io/delta/kernel/internal/snapshot/SnapshotManager.java

@@ -31,6 +31,7 @@
 import io.delta.kernel.*;
 import io.delta.kernel.engine.Engine;
 import io.delta.kernel.exceptions.CheckpointAlreadyExistsException;
+import io.delta.kernel.exceptions.InvalidTableException;
 import io.delta.kernel.exceptions.TableNotFoundException;
 import io.delta.kernel.utils.CloseableIterator;
 import io.delta.kernel.utils.FileStatus;
@@ -79,15 +80,19 @@ public SnapshotManager(Path logPath, Path tablePath) {
     public static void verifyDeltaVersions(
             List<Long> versions,
             Optional<Long> expectedStartVersion,
-            Optional<Long> expectedEndVersion) {
+            Optional<Long> expectedEndVersion,
+            Path tablePath) {
         if (!versions.isEmpty()) {
             List<Long> contVersions = LongStream
                     .rangeClosed(versions.get(0), versions.get(versions.size() -1))
                     .boxed()
                     .collect(Collectors.toList());
             if (!contVersions.equals(versions)) {
-                throw new IllegalStateException(
-                        format("Versions (%s) are not continuous", versions));
+                throw new InvalidTableException(
+                    tablePath.toString(),
+                    String.format(
+                        "Missing delta files: versions are not continuous: (%s)", versions)
+                );
             }
         }
         expectedStartVersion.ifPresent(v -> {
@@ -628,11 +633,13 @@ protected Optional<LogSegment> getLogSegmentForVersion(
         final long newVersion = deltaVersionsAfterCheckpoint.isEmpty() ?
             newCheckpointOpt.get().version : deltaVersionsAfterCheckpoint.getLast();
 
-        // In the case where `deltasAfterCheckpoint` is empty, `deltas` should still not be empty,
-        // they may just be before the checkpoint version unless we have a bug in log cleanup.
-        if (deltas.isEmpty()) {
-            throw new IllegalStateException(
-                format("Could not find any delta files for version %s", newVersion)
+        // There should be a delta file present for the newVersion that we are loading
+        // (Even if `deltasAfterCheckpoint` is empty, `deltas` should not be)
+        if (deltas.isEmpty() ||
+            FileNames.deltaVersion(deltas.get(deltas.size() - 1).getPath()) < newVersion) {
+            throw new InvalidTableException(
+                tablePath.toString(),
+                String.format("Missing delta file for version %s", newVersion)
             );
         }
 
@@ -642,19 +649,24 @@ protected Optional<LogSegment> getLogSegmentForVersion(
 
         // We may just be getting a checkpoint file after the filtering
         if (!deltaVersionsAfterCheckpoint.isEmpty()) {
+            // If we have deltas after the checkpoint, the first file should be 1 greater than our
+            // last checkpoint version. If no checkpoint is present, this means the first delta file
+            // should be version 0.
             if (deltaVersionsAfterCheckpoint.getFirst() != newCheckpointVersion + 1) {
-                throw new RuntimeException(
-                    format(
-                        "Log file not found.\nExpected: %s\nFound: %s",
-                        FileNames.deltaFile(logPath, newCheckpointVersion + 1),
-                        FileNames.deltaFile(logPath, deltaVersionsAfterCheckpoint.get(0))
+                throw new InvalidTableException(
+                    tablePath.toString(),
+                    String.format(
+                        "Unable to reconstruct table state: missing log file for version %s",
+                        newCheckpointVersion + 1
                     )
                 );
             }
             verifyDeltaVersions(
                 deltaVersionsAfterCheckpoint,
                 Optional.of(newCheckpointVersion + 1),
-                versionToLoadOpt);
+                versionToLoadOpt,
+                tablePath
+            );
         }
 
         final long lastCommitTimestamp = deltas.get(deltas.size() - 1).getModificationTime();

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

@@ -20,9 +20,11 @@ import java.util.{Arrays, Collections, Optional}
 import scala.collection.JavaConverters._
 import scala.reflect.ClassTag
 
-import io.delta.kernel.data.{ColumnVector, ColumnarBatch}
+import io.delta.kernel.data.{ColumnarBatch, ColumnVector}
+import io.delta.kernel.exceptions.InvalidTableException
 import io.delta.kernel.expressions.Predicate
 import io.delta.kernel.internal.checkpoints.{CheckpointInstance, SidecarFile}
+import io.delta.kernel.internal.fs.Path
 import io.delta.kernel.internal.snapshot.{LogSegment, SnapshotManager}
 import io.delta.kernel.internal.util.{FileNames, Utils}
 import io.delta.kernel.test.{BaseMockJsonHandler, BaseMockParquetHandler, MockFileSystemClientUtils, VectorTestUtils}
@@ -37,70 +39,82 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
     SnapshotManager.verifyDeltaVersions(
       Collections.emptyList(),
       Optional.empty(),
-      Optional.empty())
+      Optional.empty(),
+      new Path("/path/to/table"))
     // contiguous versions
     SnapshotManager.verifyDeltaVersions(
       Arrays.asList(1, 2, 3),
       Optional.empty(),
-      Optional.empty())
+      Optional.empty(),
+      new Path("/path/to/table"))
     // contiguous versions with correct `expectedStartVersion` and `expectedStartVersion`
     SnapshotManager.verifyDeltaVersions(
       Arrays.asList(1, 2, 3),
       Optional.empty(),
-      Optional.of(3))
+      Optional.of(3),
+      new Path("/path/to/table"))
     SnapshotManager.verifyDeltaVersions(
       Arrays.asList(1, 2, 3),
       Optional.of(1),
-      Optional.empty())
+      Optional.empty(),
+      new Path("/path/to/table"))
     SnapshotManager.verifyDeltaVersions(
       Arrays.asList(1, 2, 3),
       Optional.of(1),
-      Optional.of(3))
+      Optional.of(3),
+      new Path("/path/to/table"))
     // `expectedStartVersion` or `expectedEndVersion` doesn't match
     intercept[IllegalArgumentException] {
       SnapshotManager.verifyDeltaVersions(
         Arrays.asList(1, 2),
         Optional.of(0),
-        Optional.empty())
+        Optional.empty(),
+        new Path("/path/to/table"))
     }
     intercept[IllegalArgumentException] {
       SnapshotManager.verifyDeltaVersions(
         Arrays.asList(1, 2),
         Optional.empty(),
-        Optional.of(3))
+        Optional.of(3),
+        new Path("/path/to/table"))
     }
     intercept[IllegalArgumentException] {
       SnapshotManager.verifyDeltaVersions(
         Collections.emptyList(),
         Optional.of(0),
-        Optional.empty())
+        Optional.empty(),
+        new Path("/path/to/table"))
     }
     intercept[IllegalArgumentException] {
       SnapshotManager.verifyDeltaVersions(
         Collections.emptyList(),
         Optional.empty(),
-        Optional.of(3))
+        Optional.of(3),
+        new Path("/path/to/table"))
     }
     // non contiguous versions
-    intercept[IllegalStateException] {
+    intercept[InvalidTableException] {
       SnapshotManager.verifyDeltaVersions(
         Arrays.asList(1, 3),
         Optional.empty(),
-        Optional.empty())
+        Optional.empty(),
+        new Path("/path/to/table"))
     }
     // duplicates in versions
-    intercept[IllegalStateException] {
+    intercept[InvalidTableException] {
       SnapshotManager.verifyDeltaVersions(
         Arrays.asList(1, 2, 2, 3),
         Optional.empty(),
-        Optional.empty())
+        Optional.empty(),
+        new Path("/path/to/table"))
     }
     // unsorted versions
-    intercept[IllegalStateException] {
+    intercept[InvalidTableException] {
       SnapshotManager.verifyDeltaVersions(
         Arrays.asList(3, 2, 1),
         Optional.empty(),
-        Optional.empty())
+        Optional.empty(),
+        new Path("/path/to/table"))
     }
   }
 
@@ -631,11 +645,11 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
            Seq(Optional.empty(), Optional.of(10L)): Seq[Optional[java.lang.Long]]) {
       for (startCheckpoint <-
              Seq(Optional.empty(), Optional.of(10L)): Seq[Optional[java.lang.Long]]) {
-        testExpectedError[IllegalStateException](
+        testExpectedError[InvalidTableException](
           files = singularCheckpointFileStatuses(Seq(10L)),
           startCheckpoint = startCheckpoint,
           versionToLoad = versionToLoad,
-          expectedErrorMessageContains = "Could not find any delta files for version 10"
+          expectedErrorMessageContains = "Missing delta file for version 10"
         )
       }
     }
@@ -645,60 +659,47 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
     // checkpoint(10), 010.json, 011.json, 013.json
     val fileList = deltaFileStatuses(Seq(10L, 11L)) ++ deltaFileStatuses(Seq(13L)) ++
       singularCheckpointFileStatuses(Seq(10L))
-    testExpectedError[RuntimeException](
+    testExpectedError[InvalidTableException](
       fileList,
-      expectedErrorMessageContains = "Versions ([11, 13]) are not continuous"
+      expectedErrorMessageContains = "versions are not continuous: ([11, 13])"
     )
-    testExpectedError[RuntimeException](
+    testExpectedError[InvalidTableException](
       fileList,
       startCheckpoint = Optional.of(10),
-      expectedErrorMessageContains = "Versions ([11, 13]) are not continuous"
+      expectedErrorMessageContains = "versions are not continuous: ([11, 13])"
     )
-    testExpectedError[RuntimeException](
+    testExpectedError[InvalidTableException](
       fileList,
       versionToLoad = Optional.of(13),
-      expectedErrorMessageContains = "Versions ([11, 13]) are not continuous"
+      expectedErrorMessageContains = "versions are not continuous: ([11, 13])"
     )
   }
 
-  // TODO address the inconsistent behaviors and throw better error messages for corrupt listings?
-  //  (delta-io/delta#2283)
   test("getLogSegmentForVersion: corrupt listing 000.json...009.json + checkpoint(10)") {
     val fileList = deltaFileStatuses((0L until 10L)) ++ singularCheckpointFileStatuses(Seq(10L))
 
     /* ----------  version to load is 15 (greater than latest checkpoint/delta file) ---------- */
-    testExpectedError[RuntimeException](
+    testExpectedError[InvalidTableException](
       fileList,
       versionToLoad = Optional.of(15),
-      expectedErrorMessageContains = "Could not find any delta files for version 10"
+      expectedErrorMessageContains = "Missing delta file for version 10"
     )
-    testExpectedError[IllegalStateException](
+    testExpectedError[InvalidTableException](
       fileList,
       startCheckpoint = Optional.of(10),
       versionToLoad = Optional.of(15),
-      expectedErrorMessageContains = "Could not find any delta files for version 10"
+      expectedErrorMessageContains = "Missing delta file for version 10"
     )
 
     /* ---------- versionToLoad is latest (10) ---------- */
-    // (?) fails when startCheckpoint is provided, passes when it's not
-    testExpectedError[IllegalStateException](
+    testExpectedError[InvalidTableException](
       fileList,
       startCheckpoint = Optional.of(10),
-      expectedErrorMessageContains = "Could not find any delta files for version 10"
-    )
-    val logSegment = snapshotManager.getLogSegmentForVersion(
-      createMockFSListFromEngine(fileList),
-      Optional.empty(),
-      Optional.empty()
+      expectedErrorMessageContains = "Missing delta file for version 10"
     )
-    assert(logSegment.isPresent())
-    checkLogSegment(
-      logSegment.get(),
-      10,
-      Seq.empty,
-      singularCheckpointFileStatuses(Seq(10L)),
-      Some(10),
-      90 // is the last available delta file
+    testExpectedError[InvalidTableException](
+      fileList,
+      expectedErrorMessageContains = "Missing delta file for version 10"
     )
   }
 
@@ -718,41 +719,36 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
     )
     /* ---------- versionToLoad is 10 ---------- */
     // (?) throws an error
-    testExpectedError[IllegalStateException](
+    testExpectedError[InvalidTableException](
       fileList,
       versionToLoad = Optional.of(10),
-      expectedErrorMessageContains = "Could not find any delta files for version 10"
+      expectedErrorMessageContains = "Missing delta file for version 10"
     )
-    testExpectedError[IllegalStateException](
+    testExpectedError[InvalidTableException](
       fileList,
       startCheckpoint = Optional.of(10),
       versionToLoad = Optional.of(10),
-      expectedErrorMessageContains = "Could not find any delta files for version 10"
+      expectedErrorMessageContains = "Missing delta file for version 10"
     )
   }
 
   test("getLogSegmentForVersion: corrupted log missing json files / no way to construct history") {
-    def expectedErrorMessage(v: Int): String = {
-      s"""|Log file not found.
-          |Expected: ${FileNames.deltaFile(logPath, 0)}
-          |Found: ${FileNames.deltaFile(logPath, v)}""".stripMargin
-    }
-    testExpectedError[RuntimeException](
+    testExpectedError[InvalidTableException](
       deltaFileStatuses(1L until 10L),
-      expectedErrorMessageContains = expectedErrorMessage(1)
+      expectedErrorMessageContains = "missing log file for version 0"
     )
-    testExpectedError[RuntimeException](
+    testExpectedError[InvalidTableException](
       deltaFileStatuses(15L until 25L) ++ singularCheckpointFileStatuses(Seq(20L)),
       versionToLoad = Optional.of(17),
-      expectedErrorMessageContains = expectedErrorMessage(15)
+      expectedErrorMessageContains = "missing log file for version 0"
     )
-    testExpectedError[RuntimeException](
+    testExpectedError[InvalidTableException](
       deltaFileStatuses(15L until 25L) ++ singularCheckpointFileStatuses(Seq(20L)),
       startCheckpoint = Optional.of(20),
       versionToLoad = Optional.of(17),
-      expectedErrorMessageContains = expectedErrorMessage(15)
+      expectedErrorMessageContains = "missing log file for version 0"
     )
-    testExpectedError[RuntimeException](
+    testExpectedError[InvalidTableException](
       deltaFileStatuses((0L until 5L) ++ (6L until 9L)),
       expectedErrorMessageContains = "are not continuous"
     )
@@ -761,11 +757,11 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
       .map(p => FileStatus.of(p.toString, 10, 10))
       .take(4)
     val deltas = deltaFileStatuses(10L to 13L)
-    testExpectedError[RuntimeException](
+    testExpectedError[InvalidTableException](
       corruptedCheckpointStatuses ++ deltas,
       Optional.empty(),
       Optional.empty(),
-      expectedErrorMessageContains = expectedErrorMessage(10)
+      expectedErrorMessageContains = "missing log file for version 0"
     )
   }