Core: Fix REST catalog handling when the service has no view support (apache#9853)

* Core: Fix REST catalog handling when the service has no view support.

* Fix error message.

* Improve the error message.

(cherry picked from commit 1a4f23b)

Author: rdblue

core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java

@@ -59,6 +59,7 @@
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.exceptions.NoSuchViewException;
+import org.apache.iceberg.exceptions.RESTException;
 import org.apache.iceberg.hadoop.Configurable;
 import org.apache.iceberg.io.CloseableGroup;
 import org.apache.iceberg.io.FileIO;
@@ -678,8 +679,15 @@ public Transaction createTransaction() {
 
     @Override
     public Transaction replaceTransaction() {
-      if (viewExists(context, ident)) {
-        throw new AlreadyExistsException("View with same name already exists: %s", ident);
+      try {
+        if (viewExists(context, ident)) {
+          throw new AlreadyExistsException("View with same name already exists: %s", ident);
+        }
+      } catch (RESTException | UnsupportedOperationException e) {
+        // don't fail if the server doesn't support views, which could be due to:
+        // 1. server or backing catalog doesn't support views
+        // 2. newer client talks to an older server that doesn't support views
+        LOG.debug("Failed to check whether view {} exists", ident, e);
       }
 
       LoadTableResponse response = loadInternal(context, ident, snapshotMode);
@@ -974,12 +982,22 @@ public List<TableIdentifier> listViews(SessionContext context, Namespace namespa
   public View loadView(SessionContext context, TableIdentifier identifier) {
     checkViewIdentifierIsValid(identifier);
 
-    LoadViewResponse response =
-        client.get(
-            paths.view(identifier),
-            LoadViewResponse.class,
-            headers(context),
-            ErrorHandlers.viewErrorHandler());
+    LoadViewResponse response;
+    try {
+      response =
+          client.get(
+              paths.view(identifier),
+              LoadViewResponse.class,
+              headers(context),
+              ErrorHandlers.viewErrorHandler());
+    } catch (UnsupportedOperationException | RESTException e) {
+      // Normally, copying an exception message is a bad practice but engines may show just the
+      // message and suppress the exception cause when the view does not exist. Since 401 and 403
+      // responses can trigger this case, including the message increases the chances that the "Not
+      // authorized" or "Forbidden" message is preserved and shown.
+      throw new NoSuchViewException(
+          e, "Unable to load view %s.%s: %s", name(), identifier, e.getMessage());
+    }
 
     AuthSession session = tableSession(response.config(), session(context));
     ViewMetadata metadata = response.metadata();

spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java

@@ -57,7 +57,6 @@
 import org.apache.iceberg.relocated.com.google.common.util.concurrent.MoreExecutors;
 import org.apache.iceberg.spark.Spark3Util;
 import org.apache.iceberg.spark.data.TestHelpers;
-import org.apache.spark.SparkException;
 import org.apache.spark.sql.AnalysisException;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Encoders;
@@ -928,7 +927,6 @@ public synchronized void testDeleteWithSerializableIsolation() throws Interrupte
       Assert.fail("Expected a validation exception");
     } catch (ExecutionException e) {
       Throwable sparkException = e.getCause();
-      Assert.assertThat(sparkException, CoreMatchers.instanceOf(SparkException.class));
       Throwable validationException = sparkException.getCause();
       Assert.assertThat(validationException, CoreMatchers.instanceOf(ValidationException.class));
       String errMsg = validationException.getMessage();