[Kernel] Refactor all user-facing exceptions to be "KernelExceptions" (#3014)

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

In order to provide a unified and user-friendly experience we are making
all user-facing exceptions instances of one parent class
`KernelException`. This PR also attempts to unify/improve some of these
error messsages.

## How was this patch tested?

Existing tests should suffice (updated to match the changes here).

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

Yes, changes exception types + messages.

(cherry picked from commit c426f03)

Author: allisonport-db

connectors/flink/src/main/scala/io/delta/flink/internal/KernelDeltaLogDelegator.scala

@@ -20,9 +20,9 @@ package io.delta.standalone.internal
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
-
-import io.delta.kernel.{Table, TableNotFoundException}
+import io.delta.kernel.Table
 import io.delta.kernel.defaults.engine.DefaultEngine
+import io.delta.kernel.exceptions.TableNotFoundException
 import io.delta.kernel.internal.{TableImpl, SnapshotImpl => SnapshotImplKernel}
 import io.delta.standalone.VersionLog
 import io.delta.standalone.actions.{CommitInfo => CommitInfoJ}

kernel/examples/table-reader/src/main/java/io/delta/kernel/examples/BaseTableReader.java

@@ -36,8 +36,8 @@
 import org.apache.commons.cli.ParseException;
 import org.apache.hadoop.conf.Configuration;
 
-import io.delta.kernel.TableNotFoundException;
 import io.delta.kernel.engine.Engine;
+import io.delta.kernel.exceptions.TableNotFoundException;
 import io.delta.kernel.data.FilteredColumnarBatch;
 import io.delta.kernel.data.Row;
 import io.delta.kernel.expressions.Predicate;

kernel/examples/table-reader/src/main/java/io/delta/kernel/examples/MultiThreadedTableReader.java

@@ -31,6 +31,7 @@
 import io.delta.kernel.data.FilteredColumnarBatch;
 import io.delta.kernel.data.Row;
 import io.delta.kernel.examples.utils.RowSerDe;
+import io.delta.kernel.exceptions.TableNotFoundException;
 import io.delta.kernel.expressions.Predicate;
 import io.delta.kernel.types.StructType;
 import io.delta.kernel.utils.CloseableIterator;

kernel/examples/table-reader/src/main/java/io/delta/kernel/examples/SingleThreadedTableReader.java

@@ -25,6 +25,7 @@
 import io.delta.kernel.data.ColumnarBatch;
 import io.delta.kernel.data.FilteredColumnarBatch;
 import io.delta.kernel.data.Row;
+import io.delta.kernel.exceptions.TableNotFoundException;
 import io.delta.kernel.expressions.Predicate;
 import io.delta.kernel.types.StructType;
 import io.delta.kernel.utils.CloseableIterator;

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

@@ -19,7 +19,9 @@
 
 import io.delta.kernel.annotation.Evolving;
 import io.delta.kernel.engine.Engine;
-
+import io.delta.kernel.exceptions.CheckpointAlreadyExistsException;
+import io.delta.kernel.exceptions.KernelException;
+import io.delta.kernel.exceptions.TableNotFoundException;
 import io.delta.kernel.internal.TableImpl;
 
 /**
@@ -84,6 +86,8 @@ Snapshot getLatestSnapshot(Engine engine)
      * @param engine {@link Engine} instance to use in Delta Kernel.
      * @param versionId snapshot version to retrieve
      * @return an instance of {@link Snapshot}
+     * @throws KernelException if the provided version is less than the first available version
+     *                         or greater than the last available version
      * @since 3.2.0
      */
     Snapshot getSnapshotAsOfVersion(Engine engine, long versionId)
@@ -109,6 +113,8 @@ Snapshot getSnapshotAsOfVersion(Engine engine, long versionId)
      * @param millisSinceEpochUTC timestamp to fetch the snapshot for in milliseconds since the
      *                            unix epoch
      * @return an instance of {@link Snapshot}
+     * @throws KernelException if the provided timestamp is before the earliest available version or
+     *                         after the latest available version
      * @since 3.2.0
      */
     Snapshot getSnapshotAsOfTimestamp(Engine engine, long millisSinceEpochUTC)

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

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.delta.kernel;
+package io.delta.kernel.exceptions;
 
 import static java.lang.String.format;
 
@@ -26,7 +26,7 @@
  * @since 3.2.0
  */
 @Evolving
-public class CheckpointAlreadyExistsException extends IllegalArgumentException {
+public class CheckpointAlreadyExistsException extends KernelException {
     public CheckpointAlreadyExistsException(long version) {
         super(format("Checkpoint for given version %d already exists in the table", version));
     }

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

@@ -0,0 +1,39 @@
+/*
+ * 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 Kernel cannot execute the requested operation due to the operation being invalid
+ * or unsupported.
+ */
+public class KernelException extends RuntimeException {
+
+    public KernelException() {
+        super();
+    }
+
+    public KernelException(String message) {
+        super(message);
+    }
+
+    public KernelException(Throwable cause) {
+        super(cause);
+    }
+
+    public KernelException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package io.delta.kernel;
+package io.delta.kernel.exceptions;
 
 import io.delta.kernel.annotation.Evolving;
 
@@ -24,22 +24,23 @@
  * @since 3.0.0
  */
 @Evolving
-public class TableNotFoundException
-    extends Exception {
+public class TableNotFoundException extends KernelException {
 
     private final String tablePath;
 
     public TableNotFoundException(String tablePath) {
-        this.tablePath = tablePath;
+        this(tablePath, null);
     }
 
     public TableNotFoundException(String tablePath, Throwable cause) {
-        super(cause);
+        super(String.format("Delta table at path `%s` is not found", tablePath), cause);
         this.tablePath = tablePath;
     }
 
-    @Override
-    public String getMessage() {
-        return String.format("Delta table at path `%s` is not found", tablePath);
+    /**
+     * @return the provided path where no Delta table was found
+     */
+    public String getTablePath() {
+        return tablePath;
     }
 }

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

@@ -16,120 +16,122 @@
 package io.delta.kernel.internal;
 
 import java.sql.Timestamp;
-import java.util.Optional;
 
-import io.delta.kernel.expressions.Expression;
+import io.delta.kernel.exceptions.KernelException;
 
+/**
+ * Contains methods to create user-facing Delta exceptions.
+ */
 public final class DeltaErrors {
     private DeltaErrors() {}
 
-    // TODO update to be user-facing exception with future exception framework
-    //  (see delta-io/delta#2231) & document in method docs as needed (Table::getSnapshotAtVersion)
-    public static RuntimeException nonReconstructableStateException(
-            String tablePath, long version) {
+    public static KernelException versionBeforeFirstAvailableCommit(
+            String tablePath, long versionToLoad, long earliestVersion) {
         String message = String.format(
-            "%s: Unable to reconstruct state at version %s as the transaction log has been " +
-                "truncated due to manual deletion or the log retention policy and checkpoint " +
-                "retention policy.",
+            "%s: Cannot load table version %s as the transaction log has been truncated due to " +
+                "manual deletion or the log/checkpoint retention policy. The earliest available " +
+                "version is %s.",
             tablePath,
-            version);
-        return new RuntimeException(message);
+            versionToLoad,
+            earliestVersion);
+        return new KernelException(message);
     }
 
-    // TODO update to be user-facing exception with future exception framework
-    //  (see delta-io/delta#2231) & document in method docs as needed (Table::getSnapshotAtVersion)
-    public static RuntimeException nonExistentVersionException(
+    public static KernelException versionAfterLatestCommit(
             String tablePath, long versionToLoad, long latestVersion) {
         String message = String.format(
-            "%s: Trying to load a non-existent version %s. The latest version available is %s",
+            "%s: Cannot load table version %s as it does not exist. " +
+                "The latest available version is %s.",
             tablePath,
             versionToLoad,
             latestVersion);
-        return new RuntimeException(message);
+        return new KernelException(message);
     }
 
-    // TODO update to be user-facing exception with future exception framework
-    //  (see delta-io/delta#2231) & document in method docs as needed
-    //  (Table::getSnapshotAtTimestamp)
-    public static RuntimeException timestampEarlierThanTableFirstCommitException(
-            String tablePath, long providedTimestamp, long commitTimestamp) {
+    public static KernelException timestampBeforeFirstAvailableCommit(
+            String tablePath,
+            long providedTimestamp,
+            long earliestCommitTimestamp,
+            long earliestCommitVersion) {
         String message = String.format(
-            "%s: The provided timestamp %s ms (%s) is before the earliest version available. " +
-                "Please use a timestamp greater than or equal to %s ms (%s)",
+            "%s: The provided timestamp %s ms (%s) is before the earliest available version %s. " +
+                "Please use a timestamp greater than or equal to %s ms (%s).",
             tablePath,
             providedTimestamp,
             formatTimestamp(providedTimestamp),
-            commitTimestamp,
-            formatTimestamp(commitTimestamp));
-        return new RuntimeException(message);
+            earliestCommitVersion,
+            earliestCommitTimestamp,
+            formatTimestamp(earliestCommitTimestamp));
+        return new KernelException(message);
     }
 
-    // TODO update to be user-facing exception with future exception framework
-    //  (see delta-io/delta#2231) & document in method docs as needed
-    //  (Table::getSnapshotAtTimestamp)
-    public static RuntimeException timestampLaterThanTableLastCommit(
-            String tablePath, long providedTimestamp, long commitTimestamp, long commitVersion) {
-        String commitTimestampStr = formatTimestamp(commitTimestamp);
+    public static KernelException timestampAfterLatestCommit(
+            String tablePath,
+            long providedTimestamp,
+            long latestCommitTimestamp,
+            long latestCommitVersion) {
         String message = String.format(
-            "%s: The provided timestamp %s ms (%s) is after the latest commit with " +
-                "timestamp %s ms (%s). If you wish to query this version of the table please " +
-                "either provide the version %s or use the exact timestamp of the last " +
-                "commit %s ms (%s)",
+            "%s: The provided timestamp %s ms (%s) is after the latest available version %s. " +
+                "Please use a timestamp less than or equal to %s ms (%s).",
             tablePath,
             providedTimestamp,
             formatTimestamp(providedTimestamp),
-            commitTimestamp,
-            commitTimestampStr,
-            commitVersion,
-            commitTimestamp,
-            commitTimestampStr);
-        return new RuntimeException(message);
+            latestCommitVersion,
+            latestCommitTimestamp,
+            formatTimestamp(latestCommitTimestamp));
+        return new KernelException(message);
     }
 
-    // TODO: Change the exception to proper type as part of the exception framework
-    // (see delta-io/delta#2231)
-    /**
-     * Exception thrown when the expression evaluator doesn't support the given expression.
-     * @param expression
-     * @param reason Optional additional reason for why the expression is not supported.
-     * @return
-     */
-    public static UnsupportedOperationException unsupportedExpression(
-            Expression expression,
-            Optional<String> reason) {
+    /* ------------------------ PROTOCOL EXCEPTIONS ----------------------------- */
+
+    public static KernelException unsupportedReaderProtocol(
+            String tablePath, int tableReaderVersion) {
         String message = String.format(
-            "Expression evaluator doesn't support the expression: %s.%s",
-                expression,
-                reason.map(r -> " Reason: " + r).orElse(""));
-        return new UnsupportedOperationException(message);
+            "Unsupported Delta protocol reader version: table `%s` requires reader version %s " +
+                "which is unsupported by this version of Delta Kernel.",
+            tablePath,
+            tableReaderVersion);
+        return new KernelException(message);
     }
 
-    public static UnsupportedOperationException unsupportedReaderProtocol(int readVersion) {
-        throw new UnsupportedOperationException(
-                "Unsupported reader protocol version: " + readVersion);
+    public static KernelException unsupportedReaderFeature(
+            String tablePath, String readerFeature) {
+        String message = String.format(
+            "Unsupported Delta reader feature: table `%s` requires reader table feature \"%s\" " +
+                "which is unsupported by this version of Delta Kernel.",
+            tablePath,
+            readerFeature);
+        return new KernelException(message);
     }
 
-    public static UnsupportedOperationException unsupportedReadFeature(
-            int readProtocolVersion,
-            String readFeature) {
-        throw new UnsupportedOperationException(String.format(
-                "Unsupported reader protocol version: %s with feature: %s",
-                    readProtocolVersion, readFeature));
+    public static KernelException unsupportedWriterProtocol(
+            String tablePath, int tableWriterVersion) {
+        String message = String.format(
+            "Unsupported Delta protocol writer version: table `%s` requires writer version %s " +
+                "which is unsupported by this version of Delta Kernel.",
+            tablePath,
+            tableWriterVersion);
+        return new KernelException(message);
     }
 
-    public static UnsupportedOperationException unsupportedWriterProtocol(int writeVersion) {
-        throw new UnsupportedOperationException(
-                "Unsupported writer protocol version: " + writeVersion);
+    public static KernelException unsupportedWriterFeature(
+            String tablePath, String writerFeature) {
+        String message = String.format(
+            "Unsupported Delta writer feature: table `%s` requires writer table feature \"%s\" " +
+                "which is unsupported by this version of Delta Kernel.",
+            tablePath,
+            writerFeature);
+        return new KernelException(message);
     }
 
-    public static UnsupportedOperationException unsupportedWriteFeature(
-            int writeProtocolVersion,
-            String writeFeature) {
-        throw new UnsupportedOperationException(String.format(
-                "Unsupported writer protocol version: %s with feature: %s",
-                writeProtocolVersion, writeFeature));
+    public static KernelException columnInvariantsNotSupported() {
+        String message = "This version of Delta Kernel does not support writing to tables with " +
+            "column invariants present.";
+        return new KernelException(message);
     }
 
+    /* ------------------------ HELPER METHODS ----------------------------- */
+
     private static String formatTimestamp(long millisSinceEpochUTC) {
         return new Timestamp(millisSinceEpochUTC).toInstant().toString();
     }