[DO NOT MERGE][Kernel] Option 1 for wrapping all calls into the client implementation

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

@@ -0,0 +1,34 @@
+/*
+ * 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;
+
+/**
+ * TODO
+ */
+public class KernelEngineException extends RuntimeException {
+
+    public KernelEngineException(String attemptedOperation, Throwable cause) {
+        super(
+            String.format(
+                "Encountered an error from the underlying engine implementation while trying " +
+                    "to %s: %s",
+                attemptedOperation,
+                cause.getMessage()
+            ),
+            cause
+        );
+    }
+}

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

@@ -16,7 +16,9 @@
 package io.delta.kernel.internal;
 
 import java.sql.Timestamp;
+import java.util.concurrent.Callable;
 
+import io.delta.kernel.exceptions.KernelEngineException;
 import io.delta.kernel.exceptions.KernelException;
 
 /**
@@ -135,4 +137,13 @@ public static KernelException columnInvariantsNotSupported() {
     private static String formatTimestamp(long millisSinceEpochUTC) {
         return new Timestamp(millisSinceEpochUTC).toInstant().toString();
     }
+
+    // TODO add string formatting with args to avoid string formatting unless needed
+    public static <T> T wrapWithEngineException(Callable<T> s, String operation) {
+        try {
+            return s.call();
+        } catch (Exception e) {
+            throw new KernelEngineException(operation, e);
+        }
+    }
 }

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

@@ -37,6 +37,7 @@
 import io.delta.kernel.internal.skipping.DataSkippingPredicate;
 import io.delta.kernel.internal.skipping.DataSkippingUtils;
 import io.delta.kernel.internal.util.*;
+import static io.delta.kernel.internal.DeltaErrors.wrapWithEngineException;
 import static io.delta.kernel.internal.skipping.StatsSchemaHelper.getStatsSchema;
 import static io.delta.kernel.internal.util.PartitionUtils.rewritePartitionPredicateOnCheckpointFileSchema;
 import static io.delta.kernel.internal.util.PartitionUtils.rewritePartitionPredicateOnScanFileSchema;
@@ -231,14 +232,25 @@ public boolean hasNext() {
             public FilteredColumnarBatch next() {
                 FilteredColumnarBatch next = scanFileIter.next();
                 if (predicateEvaluator == null) {
-                    predicateEvaluator =
-                        engine.getExpressionHandler().getPredicateEvaluator(
+                    predicateEvaluator = wrapWithEngineException(
+                        () -> engine.getExpressionHandler().getPredicateEvaluator(
                             next.getData().getSchema(),
-                            predicateOnScanFileBatch);
+                            predicateOnScanFileBatch),
+                        String.format(
+                            "Get the predicate evaluator for partition pruning with schema=%s and" +
+                                " filter=%s",
+                            next.getData().getSchema(),
+                            predicateOnScanFileBatch
+                        )
+                    );
                 }
-                ColumnVector newSelectionVector = predicateEvaluator.eval(
-                    next.getData(),
-                    next.getSelectionVector());
+                ColumnVector newSelectionVector = wrapWithEngineException(
+                    () -> predicateEvaluator.eval(
+                        next.getData(),
+                        next.getSelectionVector()),
+                    String.format(
+                        "Evaluating the partition expression %s", predicateOnScanFileBatch)
+                );
                 return new FilteredColumnarBatch(
                     next.getData(),
                     Optional.of(newSelectionVector));

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActiveAddFilesIterator.java

@@ -33,6 +33,7 @@
 import io.delta.kernel.internal.fs.Path;
 import io.delta.kernel.internal.replay.LogReplayUtils.UniqueFileActionTuple;
 import io.delta.kernel.internal.util.Utils;
+import static io.delta.kernel.internal.DeltaErrors.wrapWithEngineException;
 import static io.delta.kernel.internal.replay.LogReplay.ADD_FILE_DV_ORDINAL;
 import static io.delta.kernel.internal.replay.LogReplay.ADD_FILE_ORDINAL;
 import static io.delta.kernel.internal.replay.LogReplay.ADD_FILE_PATH_ORDINAL;
@@ -217,11 +218,16 @@ private void prepareNext() {
         // Step 4: TODO: remove this step. This is a temporary requirement until the path
         //         in `add` is converted to absolute path.
         if (tableRootVectorGenerator == null) {
-            tableRootVectorGenerator = engine.getExpressionHandler()
-                .getEvaluator(
-                    scanAddFiles.getSchema(),
-                    Literal.ofString(tableRoot.toUri().toString()),
-                    StringType.STRING);
+            final ColumnarBatch finalScanAddFiles = scanAddFiles;
+            tableRootVectorGenerator = wrapWithEngineException(
+                () -> engine.getExpressionHandler()
+                    .getEvaluator(
+                        finalScanAddFiles.getSchema(),
+                        Literal.ofString(tableRoot.toUri().toString()),
+                        StringType.STRING),
+                "Get the expression evaluator for the table root"
+            );
+
         }
         ColumnVector tableRootVector = tableRootVectorGenerator.eval(scanAddFiles);
         scanAddFiles = scanAddFiles.withNewColumn(