Report S3 errors in metadata access as external

Before this, we'd report S3 errors (such as 403) as GENERIC_INTERNAL_ERROR.

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/AbstractIcebergTableOperations.java

@@ -23,6 +23,7 @@
 import io.trino.metastore.StorageFormat;
 import io.trino.plugin.iceberg.IcebergExceptions;
 import io.trino.plugin.iceberg.util.HiveSchemaUtil;
+import io.trino.spi.TrinoException;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.SchemaTableName;
 import jakarta.annotation.Nullable;
@@ -34,6 +35,7 @@
 import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.types.Types.NestedField;
 
+import java.io.UncheckedIOException;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
@@ -47,6 +49,7 @@
 import static io.trino.hive.formats.HiveClassNames.FILE_INPUT_FORMAT_CLASS;
 import static io.trino.hive.formats.HiveClassNames.FILE_OUTPUT_FORMAT_CLASS;
 import static io.trino.hive.formats.HiveClassNames.LAZY_SIMPLE_SERDE_CLASS;
+import static io.trino.plugin.iceberg.IcebergErrorCode.ICEBERG_INVALID_METADATA;
 import static io.trino.plugin.iceberg.IcebergExceptions.translateMetadataException;
 import static io.trino.plugin.iceberg.IcebergTableName.isMaterializedViewStorage;
 import static io.trino.plugin.iceberg.IcebergUtil.METADATA_FOLDER_NAME;
@@ -262,6 +265,9 @@ protected void refreshFromMetadataLocation(String newLocation, Function<String,
                             .build())
                     .get(() -> metadataLoader.apply(newLocation));
         }
+        catch (UncheckedIOException e) {
+            throw new TrinoException(ICEBERG_INVALID_METADATA, "Error accessing metadata file for table %s".formatted(getSchemaTableName().toString()), e);
+        }
         catch (Throwable failure) {
             throw translateMetadataException(failure, getSchemaTableName().toString());
         }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/hms/AbstractMetastoreTableOperations.java

@@ -68,7 +68,7 @@ protected AbstractMetastoreTableOperations(
     }
 
     @Override
-    protected final String getRefreshedLocation(boolean invalidateCaches)
+    protected String getRefreshedLocation(boolean invalidateCaches)
     {
         if (invalidateCaches) {
             metastore.invalidateTable(database, tableName);

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/file/TestAbstractIcebergTableOperations.java

@@ -0,0 +1,76 @@
+/*
+ * 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.trino.plugin.iceberg.catalog.file;
+
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.filesystem.local.LocalFileSystemFactory;
+import io.trino.metastore.HiveMetastore;
+import io.trino.plugin.iceberg.fileio.ForwardingFileIo;
+import org.apache.iceberg.io.InputFile;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Optional;
+
+import static io.trino.plugin.hive.metastore.cache.CachingHiveMetastore.createPerTransactionCache;
+import static io.trino.plugin.hive.metastore.file.TestingFileHiveMetastore.createTestingFileHiveMetastore;
+import static io.trino.plugin.iceberg.IcebergErrorCode.ICEBERG_INVALID_METADATA;
+import static io.trino.testing.TestingConnectorSession.SESSION;
+import static io.trino.testing.assertions.TrinoExceptionAssert.assertTrinoExceptionThrownBy;
+
+public class TestAbstractIcebergTableOperations
+{
+    @Test
+    public void testS3ErrorReporting()
+            throws IOException
+    {
+        Path tempDir = Files.createTempDirectory("test_s3_error_reporting");
+        File metastoreDir = tempDir.resolve("iceberg_data").toFile();
+        metastoreDir.mkdirs();
+        TrinoFileSystemFactory fileSystemFactory = new LocalFileSystemFactory(metastoreDir.toPath());
+        HiveMetastore metastore = createTestingFileHiveMetastore(fileSystemFactory, Location.of("local:///"));
+
+        FileMetastoreTableOperations fileMetastoreTableOperations = new FileMetastoreTableOperations(
+                new ForwardingFileIo(fileSystemFactory.create(SESSION))
+                {
+                    @Override
+                    public InputFile newInputFile(String path)
+                    {
+                        // Mimic ForwardingInputFile.newStream() behavior when there's an S3 error.
+                        throw new UncheckedIOException(new IOException());
+                    }
+                },
+                createPerTransactionCache(metastore, 1000),
+                SESSION,
+                "test-database",
+                "test-table",
+                Optional.of("test-owner"),
+                Optional.empty())
+        {
+            // Without this, we'd have to create a table that's never accessed anyway, because we're simulating S3 errors.
+            @Override
+            protected String getRefreshedLocation(boolean invalidateCaches)
+            {
+                return "local:///0.metadata.json";
+            }
+        };
+
+        assertTrinoExceptionThrownBy(fileMetastoreTableOperations::refresh).hasErrorCode(ICEBERG_INVALID_METADATA);
+    }
+}