Repository: trinodb/trino
Issue: Correctness issue with double slash location on Glue & S3 in Hive connector

The file is generated correctly under the table location, but it returns an empty result. Delta and Iceberg connectors are fine as far as I confirmed. 
```sql
create table test_slash (c1 int) with (external_location = 's3://bucket/foo//bar');
insert into test_slash values (1);
table test_slash;
 c1
----
(0 rows)
```

Changed files:
plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/TestSemiTransactionalHiveMetastore.java
plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java
plugin/trino-hive/src/test/java/io/trino/plugin/hive/AbstractTestHiveLocal.java
plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/metastore/glue/TestDeltaLakeConcurrentModificationGlueMetastore.java
plugin/trino-hive/src/main/java/io/trino/plugin/hive/PartitionUpdate.java
plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastore.java
plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/SyncPartitionMetadataProcedure.java
plugin/trino-hive/src/main/java/io/trino/plugin/hive/BackgroundHiveSplitLoader.java
plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveCreateExternalTable.java
plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestHiveGlueMetastore.java
plugin/trino-hive/src/test/java/io/trino/plugin/hive/AbstractTestHive.java
plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/SemiTransactionalHiveMetastore.java
plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestPartitionUpdate.java
plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriter.java
plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/RegisterPartitionProcedure.java
plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveS3AndGlueMetastoreTest.java
plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveLocationService.java

diff --git a/plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/metastore/glue/TestDeltaLakeConcurrentModificationGlueMetastore.java b/plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/metastore/glue/TestDeltaLakeConcurrentModificationGlueMetastore.java
index f4aaad4880..de99991545 100644
--- a/plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/metastore/glue/TestDeltaLakeConcurrentModificationGlueMetastore.java
+++ b/plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/metastore/glue/TestDeltaLakeConcurrentModificationGlueMetastore.java
@@ -17,6 +17,8 @@ import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
 import com.amazonaws.services.glue.AWSGlueAsync;
 import com.amazonaws.services.glue.model.ConcurrentModificationException;
 import io.trino.Session;
+import io.trino.filesystem.hdfs.HdfsFileSystemFactory;
+import io.trino.hdfs.TrinoHdfsFileSystemStats;
 import io.trino.plugin.deltalake.TestingDeltaLakePlugin;
 import io.trino.plugin.deltalake.metastore.TestingDeltaLakeMetastoreModule;
 import io.trino.plugin.hive.metastore.glue.DefaultGlueColumnStatisticsProviderFactory;
@@ -92,6 +94,7 @@ public class TestDeltaLakeConcurrentModificationGlueMetastore
         });
 
         metastore = new GlueHiveMetastore(
+                new HdfsFileSystemFactory(HDFS_ENVIRONMENT, new TrinoHdfsFileSystemStats()),
                 HDFS_ENVIRONMENT,
                 glueConfig,
                 directExecutor(),
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/BackgroundHiveSplitLoader.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/BackgroundHiveSplitLoader.java
index 1da5e9d99c..354fc61f27 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/BackgroundHiveSplitLoader.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/BackgroundHiveSplitLoader.java
@@ -106,6 +106,7 @@ import static com.google.common.util.concurrent.Futures.immediateVoidFuture;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
 import static io.airlift.concurrent.MoreFutures.addExceptionCallback;
 import static io.airlift.concurrent.MoreFutures.toListenableFuture;
+import static io.trino.filesystem.hdfs.HadoopPaths.hadoopPath;
 import static io.trino.hdfs.ConfigurationUtils.toJobConf;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_BAD_DATA;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_EXCEEDED_PARTITION_LIMIT;
@@ -429,7 +430,8 @@ public class BackgroundHiveSplitLoader
             return COMPLETED_FUTURE;
         }
 
-        Path path = new Path(getPartitionLocation(table, partition.getPartition()));
+        Location location = Location.of(getPartitionLocation(table, partition.getPartition()));
+        Path path = hadoopPath(location);
         Configuration configuration = hdfsEnvironment.getConfiguration(hdfsContext, path);
         InputFormat<?, ?> inputFormat = getInputFormat(configuration, schema, false);
         FileSystem fs = hdfsEnvironment.getFileSystem(hdfsContext, path);
@@ -548,7 +550,6 @@ public class BackgroundHiveSplitLoader
         }
 
         TrinoFileSystem trinoFileSystem = fileSystemFactory.create(session);
-        Location location = Location.of(path.toString());
         // Bucketed partitions are fully loaded immediately since all files must be loaded to determine the file to bucket mapping
         if (tableBucketInfo.isPresent()) {
             List<TrinoFileStatus> files = listBucketFiles(trinoFileSystem, location, splitFactory.getPartitionName());
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveLocationService.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveLocationService.java
index 6da35e45ba..e91e99e58d 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveLocationService.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveLocationService.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.fs.Path;
 
 import java.util.Optional;
 
+import static io.trino.filesystem.hdfs.HadoopPaths.hadoopPath;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_PATH_ALREADY_EXISTS;
 import static io.trino.plugin.hive.LocationHandle.WriteMode.DIRECT_TO_TARGET_EXISTING_DIRECTORY;
 import static io.trino.plugin.hive.LocationHandle.WriteMode.DIRECT_TO_TARGET_NEW_DIRECTORY;
@@ -63,7 +64,7 @@ public class HiveLocationService
         Location targetPath = getTableDefaultLocation(context, metastore, hdfsEnvironment, schemaName, tableName);
 
         // verify the target directory for table
-        if (pathExists(context, hdfsEnvironment, new Path(targetPath.toString()))) {
+        if (pathExists(context, hdfsEnvironment, hadoopPath(targetPath))) {
             throw new TrinoException(HIVE_PATH_ALREADY_EXISTS, format("Target directory for table '%s.%s' already exists: %s", schemaName, tableName, targetPath));
         }
         return targetPath;
@@ -76,13 +77,13 @@ public class HiveLocationService
         Location targetPath = externalLocation.orElseGet(() -> getTableDefaultLocation(context, metastore, hdfsEnvironment, schemaName, tableName));
 
         // verify the target directory for the table
-        if (pathExists(context, hdfsEnvironment, new Path(targetPath.toString()))) {
+        if (pathExists(context, hdfsEnvironment, hadoopPath(targetPath))) {
             throw new TrinoException(HIVE_PATH_ALREADY_EXISTS, format("Target directory for table '%s.%s' already exists: %s", schemaName, tableName, targetPath));
         }
 
         // TODO detect when existing table's location is a on a different file system than the temporary directory
-        if (shouldUseTemporaryDirectory(context, new Path(targetPath.toString()), externalLocation.isPresent())) {
-            Location writePath = createTemporaryPath(context, hdfsEnvironment, new Path(targetPath.toString()), temporaryStagingDirectoryPath);
+        if (shouldUseTemporaryDirectory(context, hadoopPath(targetPath), externalLocation.isPresent())) {
+            Location writePath = createTemporaryPath(context, hdfsEnvironment, hadoopPath(targetPath), temporaryStagingDirectoryPath);
             return new LocationHandle(targetPath, writePath, STAGE_AND_MOVE_TO_TARGET_DIRECTORY);
         }
         return new LocationHandle(targetPath, targetPath, DIRECT_TO_TARGET_NEW_DIRECTORY);
@@ -94,8 +95,8 @@ public class HiveLocationService
         HdfsContext context = new HdfsContext(session);
         Location targetPath = Location.of(table.getStorage().getLocation());
 
-        if (shouldUseTemporaryDirectory(context, new Path(targetPath.toString()), false) && !isTransactionalTable(table.getParameters())) {
-            Location writePath = createTemporaryPath(context, hdfsEnvironment, new Path(targetPath.toString()), temporaryStagingDirectoryPath);
+        if (shouldUseTemporaryDirectory(context, hadoopPath(targetPath), false) && !isTransactionalTable(table.getParameters())) {
+            Location writePath = createTemporaryPath(context, hdfsEnvironment, hadoopPath(targetPath), temporaryStagingDirectoryPath);
             return new LocationHandle(targetPath, writePath, STAGE_AND_MOVE_TO_TARGET_DIRECTORY);
         }
         return new LocationHandle(targetPath, targetPath, DIRECT_TO_TARGET_EXISTING_DIRECTORY);
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java
index 564f128dd2..cd3d0dc742 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java
@@ -165,6 +165,7 @@ import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static com.google.common.collect.Iterables.concat;
 import static com.google.common.collect.Iterables.getOnlyElement;
+import static io.trino.filesystem.hdfs.HadoopPaths.hadoopPath;
 import static io.trino.hdfs.ConfigurationUtils.toJobConf;
 import static io.trino.plugin.hive.HiveAnalyzeProperties.getColumnNames;
 import static io.trino.plugin.hive.HiveAnalyzeProperties.getPartitionList;
@@ -1734,7 +1735,7 @@ public class HiveMetadata
             partitionUpdates = PartitionUpdate.mergePartitionUpdates(concat(partitionUpdates, partitionUpdatesForMissingBuckets));
             for (PartitionUpdate partitionUpdate : partitionUpdatesForMissingBuckets) {
                 Optional<Partition> partition = table.getPartitionColumns().isEmpty() ? Optional.empty() : Optional.of(buildPartitionObject(session, table, partitionUpdate));
-                createEmptyFiles(session, partitionUpdate.getWritePath(), table, partition, partitionUpdate.getFileNames());
+                createEmptyFiles(session, hadoopPath(partitionUpdate.getWritePath()), table, partition, partitionUpdate.getFileNames());
             }
             if (handle.isTransactional()) {
                 AcidTransaction transaction = handle.getTransaction();
@@ -1765,7 +1766,7 @@ public class HiveMetadata
             tableStatistics = new PartitionStatistics(createEmptyStatistics(), ImmutableMap.of());
         }
 
-        Optional<Path> writePath = Optional.of(new Path(writeInfo.writePath().toString()));
+        Optional<Location> writePath = Optional.of(Location.of(writeInfo.writePath().toString()));
         if (handle.getPartitionedBy().isEmpty()) {
             List<String> fileNames;
             if (partitionUpdates.isEmpty()) {
@@ -1845,7 +1846,7 @@ public class HiveMetadata
     private List<String> computeFileNamesForMissingBuckets(
             ConnectorSession session,
             HiveStorageFormat storageFormat,
-            Path targetPath,
+            Location targetPath,
             int bucketCount,
             boolean transactionalCreateTable,
             PartitionUpdate partitionUpdate)
@@ -1855,7 +1856,7 @@ public class HiveMetadata
             return ImmutableList.of();
         }
         HdfsContext hdfsContext = new HdfsContext(session);
-        JobConf conf = toJobConf(hdfsEnvironment.getConfiguration(hdfsContext, targetPath));
+        JobConf conf = toJobConf(hdfsEnvironment.getConfiguration(hdfsContext, hadoopPath(targetPath)));
         configureCompression(conf, selectCompressionCodec(session, storageFormat));
         String fileExtension = HiveWriterFactory.getFileExtension(conf, fromHiveStorageFormat(storageFormat));
         Set<String> fileNames = ImmutableSet.copyOf(partitionUpdate.getFileNames());
@@ -2141,7 +2142,7 @@ public class HiveMetadata
                             statistics,
                             handle.isRetriesEnabled());
                 }
-                createEmptyFiles(session, partitionUpdate.getWritePath(), table, partition, partitionUpdate.getFileNames());
+                createEmptyFiles(session, hadoopPath(partitionUpdate.getWritePath()), table, partition, partitionUpdate.getFileNames());
             }
         }
 
@@ -2227,10 +2228,10 @@ public class HiveMetadata
                         getColumnStatistics(partitionComputedStatistics, partitionName, partitionValues, partitionTypes));
                 if (partitionUpdate.getUpdateMode() == OVERWRITE) {
                     if (handle.getLocationHandle().getWriteMode() == DIRECT_TO_TARGET_EXISTING_DIRECTORY) {
-                        removeNonCurrentQueryFiles(session, partitionUpdate.getTargetPath());
+                        removeNonCurrentQueryFiles(session, hadoopPath(partitionUpdate.getTargetPath()));
                         if (handle.isRetriesEnabled()) {
                             HdfsContext hdfsContext = new HdfsContext(session);
-                            cleanExtraOutputFiles(hdfsEnvironment, hdfsContext, session.getQueryId(), partitionUpdate.getTargetPath(), ImmutableSet.copyOf(partitionUpdate.getFileNames()));
+                            cleanExtraOutputFiles(hdfsEnvironment, hdfsContext, session.getQueryId(), hadoopPath(partitionUpdate.getTargetPath()), ImmutableSet.copyOf(partitionUpdate.getFileNames()));
                         }
                     }
                     else {
@@ -2555,22 +2556,22 @@ public class HiveMetadata
         }
 
         // path to be deleted
-        Set<Path> scannedPaths = splitSourceInfo.stream()
-                .map(file -> new Path((String) file))
+        Set<Location> scannedPaths = splitSourceInfo.stream()
+                .map(file -> Location.of((String) file))
                 .collect(toImmutableSet());
         // track remaining files to be delted for error reporting
-        Set<Path> remainingFilesToDelete = new HashSet<>(scannedPaths);
+        Set<Location> remainingFilesToDelete = new HashSet<>(scannedPaths);
 
         // delete loop
         boolean someDeleted = false;
-        Optional<Path> firstScannedPath = Optional.empty();
+        Optional<Location> firstScannedPath = Optional.empty();
         try {
-            for (Path scannedPath : scannedPaths) {
+            for (Location scannedPath : scannedPaths) {
                 if (firstScannedPath.isEmpty()) {
                     firstScannedPath = Optional.of(scannedPath);
                 }
                 retry().run("delete " + scannedPath, () -> {
-                    checkedDelete(fs, scannedPath, false);
+                    checkedDelete(fs, hadoopPath(scannedPath), false);
                     return null;
                 });
                 someDeleted = true;
@@ -2578,7 +2579,7 @@ public class HiveMetadata
             }
         }
         catch (Exception e) {
-            if (!someDeleted && (firstScannedPath.isEmpty() || exists(fs, firstScannedPath.get()))) {
+            if (!someDeleted && (firstScannedPath.isEmpty() || exists(fs, hadoopPath(firstScannedPath.get())))) {
                 // we are good - we did not delete any source files so we can just throw error and allow rollback to happend
                 // if someDeleted flag is false we do extra checkig if first file we tried to delete is still there. There is a chance that
                 // fs.delete above could throw exception but file was actually deleted.
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriter.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriter.java
index 7628b45362..4445ff3556 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriter.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriter.java
@@ -14,6 +14,7 @@
 package io.trino.plugin.hive;
 
 import com.google.common.collect.ImmutableList;
+import io.trino.filesystem.Location;
 import io.trino.plugin.hive.PartitionUpdate.UpdateMode;
 import io.trino.spi.Page;
 
@@ -114,8 +115,8 @@ public class HiveWriter
         return new PartitionUpdate(
                 partitionName.orElse(""),
                 updateMode,
-                writePath,
-                targetPath,
+                Location.of(writePath),
+                Location.of(targetPath),
                 ImmutableList.of(fileName),
                 rowCount,
                 inputSizeInBytes,
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/PartitionUpdate.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/PartitionUpdate.java
index 926d02ae35..e6b26cf332 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/PartitionUpdate.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/PartitionUpdate.java
@@ -17,8 +17,8 @@ import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Multimaps;
+import io.trino.filesystem.Location;
 import io.trino.spi.TrinoException;
-import org.apache.hadoop.fs.Path;
 
 import java.util.Collection;
 import java.util.List;
@@ -33,8 +33,8 @@ public class PartitionUpdate
 {
     private final String name;
     private final UpdateMode updateMode;
-    private final Path writePath;
-    private final Path targetPath;
+    private final Location writePath;
+    private final Location targetPath;
     private final List<String> fileNames;
     private final long rowCount;
     private final long inMemoryDataSizeInBytes;
@@ -54,8 +54,8 @@ public class PartitionUpdate
         this(
                 name,
                 updateMode,
-                new Path(requireNonNull(writePath, "writePath is null")),
-                new Path(requireNonNull(targetPath, "targetPath is null")),
+                Location.of(requireNonNull(writePath, "writePath is null")),
+                Location.of(requireNonNull(targetPath, "targetPath is null")),
                 fileNames,
                 rowCount,
                 inMemoryDataSizeInBytes,
@@ -65,8 +65,8 @@ public class PartitionUpdate
     public PartitionUpdate(
             String name,
             UpdateMode updateMode,
-            Path writePath,
-            Path targetPath,
+            Location writePath,
+            Location targetPath,
             List<String> fileNames,
             long rowCount,
             long inMemoryDataSizeInBytes,
@@ -101,12 +101,12 @@ public class PartitionUpdate
         return updateMode;
     }
 
-    public Path getWritePath()
+    public Location getWritePath()
     {
         return writePath;
     }
 
-    public Path getTargetPath()
+    public Location getTargetPath()
     {
         return targetPath;
     }
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/SemiTransactionalHiveMetastore.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/SemiTransactionalHiveMetastore.java
index 3e453c80d7..4aaf7410d6 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/SemiTransactionalHiveMetastore.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/SemiTransactionalHiveMetastore.java
@@ -98,6 +98,7 @@ import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static io.airlift.concurrent.MoreFutures.getFutureValue;
+import static io.trino.filesystem.hdfs.HadoopPaths.hadoopPath;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_CORRUPTED_COLUMN_STATISTICS;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_FILESYSTEM_ERROR;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_METASTORE_ERROR;
@@ -562,7 +563,7 @@ public class SemiTransactionalHiveMetastore
             ConnectorSession session,
             Table table,
             PrincipalPrivileges principalPrivileges,
-            Optional<Path> currentPath,
+            Optional<Location> currentPath,
             Optional<List<String>> files,
             boolean ignoreExisting,
             PartitionStatistics statistics,
@@ -680,7 +681,7 @@ public class SemiTransactionalHiveMetastore
             ConnectorSession session,
             String databaseName,
             String tableName,
-            Path currentLocation,
+            Location currentLocation,
             List<String> fileNames,
             PartitionStatistics statisticsUpdate,
             boolean cleanExtraOutputFilesOnCommit)
@@ -792,7 +793,7 @@ public class SemiTransactionalHiveMetastore
                             new TableAndMergeResults(
                                     table,
                                     Optional.of(principalPrivileges),
-                                    Optional.of(new Path(currentLocation.toString())),
+                                    Optional.of(currentLocation),
                                     partitionUpdateAndMergeResults,
                                     partitions),
                             hdfsContext,
@@ -968,7 +969,7 @@ public class SemiTransactionalHiveMetastore
             String databaseName,
             String tableName,
             Partition partition,
-            Path currentLocation,
+            Location currentLocation,
             Optional<List<String>> files,
             PartitionStatistics statistics,
             boolean cleanExtraOutputFilesOnCommit)
@@ -1733,16 +1734,16 @@ public class SemiTransactionalHiveMetastore
                 }
             }
 
-            Path currentPath = tableAndMore.getCurrentLocation()
+            Location currentPath = tableAndMore.getCurrentLocation()
                     .orElseThrow(() -> new IllegalArgumentException("location should be present for alter table"));
-            Path targetPath = new Path(targetLocation);
+            Location targetPath = Location.of(targetLocation);
             if (!targetPath.equals(currentPath)) {
                 renameDirectory(
                         hdfsContext,
                         hdfsEnvironment,
-                        currentPath,
-                        targetPath,
-                        () -> cleanUpTasksForAbort.add(new DirectoryCleanUpTask(hdfsContext, targetPath, true)));
+                        hadoopPath(currentPath),
+                        hadoopPath(targetPath),
+                        () -> cleanUpTasksForAbort.add(new DirectoryCleanUpTask(hdfsContext, hadoopPath(targetPath), true)));
             }
             // Partition alter must happen regardless of whether original and current location is the same
             // because metadata might change: e.g. storage format, column types, etc
@@ -1766,8 +1767,8 @@ public class SemiTransactionalHiveMetastore
                 Optional<String> targetLocation = table.getStorage().getOptionalLocation();
                 if (targetLocation.isPresent()) {
                     checkArgument(!targetLocation.get().isEmpty(), "target location is empty");
-                    Optional<Path> currentPath = tableAndMore.getCurrentLocation();
-                    Path targetPath = new Path(targetLocation.get());
+                    Optional<Location> currentPath = tableAndMore.getCurrentLocation();
+                    Location targetPath = Location.of(targetLocation.get());
                     if (table.getPartitionColumns().isEmpty() && currentPath.isPresent()) {
                         // CREATE TABLE AS SELECT unpartitioned table
                         if (targetPath.equals(currentPath.get())) {
@@ -1777,15 +1778,15 @@ public class SemiTransactionalHiveMetastore
                             renameDirectory(
                                     context,
                                     hdfsEnvironment,
-                                    currentPath.get(),
-                                    targetPath,
-                                    () -> cleanUpTasksForAbort.add(new DirectoryCleanUpTask(context, targetPath, true)));
+                                    hadoopPath(currentPath.get()),
+                                    hadoopPath(targetPath),
+                                    () -> cleanUpTasksForAbort.add(new DirectoryCleanUpTask(context, hadoopPath(targetPath), true)));
                         }
                     }
                     else {
                         // CREATE TABLE AS SELECT partitioned table, or
                         // CREATE TABLE partitioned/unpartitioned table (without data)
-                        if (pathExists(context, hdfsEnvironment, targetPath)) {
+                        if (pathExists(context, hdfsEnvironment, hadoopPath(targetPath))) {
                             if (currentPath.isPresent() && currentPath.get().equals(targetPath)) {
                                 // It is okay to skip directory creation when currentPath is equal to targetPath
                                 // because the directory may have been created when creating partition directories.
@@ -1799,8 +1800,8 @@ public class SemiTransactionalHiveMetastore
                             }
                         }
                         else {
-                            cleanUpTasksForAbort.add(new DirectoryCleanUpTask(context, targetPath, true));
-                            createDirectory(context, hdfsEnvironment, targetPath);
+                            cleanUpTasksForAbort.add(new DirectoryCleanUpTask(context, hadoopPath(targetPath), true));
+                            createDirectory(context, hdfsEnvironment, hadoopPath(targetPath));
                         }
                     }
                 }
@@ -1813,10 +1814,10 @@ public class SemiTransactionalHiveMetastore
         {
             deleteOnly = false;
             Table table = tableAndMore.getTable();
-            Path targetPath = new Path(table.getStorage().getLocation());
+            Location targetPath = Location.of(table.getStorage().getLocation());
             tablesToInvalidate.add(table);
-            Path currentPath = tableAndMore.getCurrentLocation().orElseThrow();
-            cleanUpTasksForAbort.add(new DirectoryCleanUpTask(context, targetPath, false));
+            Location currentPath = Location.of(tableAndMore.getCurrentLocation().orElseThrow().toString());
+            cleanUpTasksForAbort.add(new DirectoryCleanUpTask(context, hadoopPath(targetPath), false));
 
             if (!targetPath.equals(currentPath)) {
                 // if staging directory is used we cherry-pick files to be moved
@@ -1846,9 +1847,9 @@ public class SemiTransactionalHiveMetastore
 
             deleteOnly = false;
             Table table = tableAndMore.getTable();
-            Path targetPath = new Path(table.getStorage().getLocation());
-            Path currentPath = tableAndMore.getCurrentLocation().get();
-            cleanUpTasksForAbort.add(new DirectoryCleanUpTask(context, targetPath, false));
+            Location targetPath = Location.of(table.getStorage().getLocation());
+            Location currentPath = Location.of(tableAndMore.getCurrentLocation().get().toString());
+            cleanUpTasksForAbort.add(new DirectoryCleanUpTask(context, hadoopPath(targetPath), false));
             if (!targetPath.equals(currentPath)) {
                 asyncRename(hdfsEnvironment, fileSystemExecutor, fileSystemOperationsCancelled, fileSystemOperationFutures, context, currentPath, targetPath, tableAndMore.getFileNames().get());
             }
@@ -1920,15 +1921,15 @@ public class SemiTransactionalHiveMetastore
                 }
             }
 
-            Path currentPath = partitionAndMore.getCurrentLocation();
-            Path targetPath = new Path(targetLocation);
+            Location currentPath = partitionAndMore.getCurrentLocation();
+            Location targetPath = Location.of(targetLocation);
             if (!targetPath.equals(currentPath)) {
                 renameDirectory(
                         hdfsContext,
                         hdfsEnvironment,
-                        currentPath,
-                        targetPath,
-                        () -> cleanUpTasksForAbort.add(new DirectoryCleanUpTask(hdfsContext, targetPath, true)));
+                        hadoopPath(currentPath),
+                        hadoopPath(targetPath),
+                        () -> cleanUpTasksForAbort.add(new DirectoryCleanUpTask(hdfsContext, hadoopPath(targetPath), true)));
             }
             // Partition alter must happen regardless of whether original and current location is the same
             // because metadata might change: e.g. storage format, column types, etc
@@ -1944,7 +1945,7 @@ public class SemiTransactionalHiveMetastore
             }
             verify(partitionAndMore.hasFileNames(), "fileNames expected to be set if isCleanExtraOutputFilesOnCommit is true");
 
-            SemiTransactionalHiveMetastore.cleanExtraOutputFiles(hdfsEnvironment, hdfsContext, queryId, partitionAndMore.getCurrentLocation(), ImmutableSet.copyOf(partitionAndMore.getFileNames()));
+            SemiTransactionalHiveMetastore.cleanExtraOutputFiles(hdfsEnvironment, hdfsContext, queryId, hadoopPath(partitionAndMore.getCurrentLocation()), ImmutableSet.copyOf(partitionAndMore.getFileNames()));
         }
 
         private void cleanExtraOutputFiles(HdfsContext hdfsContext, String queryId, TableAndMore tableAndMore)
@@ -1952,9 +1953,9 @@ public class SemiTransactionalHiveMetastore
             if (!tableAndMore.isCleanExtraOutputFilesOnCommit()) {
                 return;
             }
-            Path tableLocation = tableAndMore.getCurrentLocation().orElseThrow(() -> new IllegalArgumentException("currentLocation expected to be set if isCleanExtraOutputFilesOnCommit is true"));
+            Location tableLocation = tableAndMore.getCurrentLocation().orElseThrow(() -> new IllegalArgumentException("currentLocation expected to be set if isCleanExtraOutputFilesOnCommit is true"));
             List<String> files = tableAndMore.getFileNames().orElseThrow(() -> new IllegalArgumentException("fileNames expected to be set if isCleanExtraOutputFilesOnCommit is true"));
-            SemiTransactionalHiveMetastore.cleanExtraOutputFiles(hdfsEnvironment, hdfsContext, queryId, tableLocation, ImmutableSet.copyOf(files));
+            SemiTransactionalHiveMetastore.cleanExtraOutputFiles(hdfsEnvironment, hdfsContext, queryId, hadoopPath(tableLocation), ImmutableSet.copyOf(files));
         }
 
         private PartitionStatistics getExistingPartitionStatistics(Partition partition, String partitionName)
@@ -1989,8 +1990,8 @@ public class SemiTransactionalHiveMetastore
 
             Partition partition = partitionAndMore.getPartition();
             String targetLocation = partition.getStorage().getLocation();
-            Path currentPath = partitionAndMore.getCurrentLocation();
-            Path targetPath = new Path(targetLocation);
+            Location currentPath = partitionAndMore.getCurrentLocation();
+            Location targetPath = Location.of(targetLocation);
 
             cleanExtraOutputFiles(hdfsContext, queryId, partitionAndMore);
 
@@ -2002,19 +2003,19 @@ public class SemiTransactionalHiveMetastore
                 if (fileSystemOperationsCancelled.get()) {
                     return;
                 }
-                if (pathExists(hdfsContext, hdfsEnvironment, currentPath)) {
+                if (pathExists(hdfsContext, hdfsEnvironment, hadoopPath(currentPath))) {
                     if (!targetPath.equals(currentPath)) {
                         renameDirectory(
                                 hdfsContext,
                                 hdfsEnvironment,
-                                currentPath,
-                                targetPath,
-                                () -> cleanUpTasksForAbort.add(new DirectoryCleanUpTask(hdfsContext, targetPath, true)));
+                                hadoopPath(currentPath),
+                                hadoopPath(targetPath),
+                                () -> cleanUpTasksForAbort.add(new DirectoryCleanUpTask(hdfsContext, hadoopPath(targetPath), true)));
                     }
                 }
                 else {
-                    cleanUpTasksForAbort.add(new DirectoryCleanUpTask(hdfsContext, targetPath, true));
-                    createDirectory(hdfsContext, hdfsEnvironment, targetPath);
+                    cleanUpTasksForAbort.add(new DirectoryCleanUpTask(hdfsContext, hadoopPath(targetPath), true));
+                    createDirectory(hdfsContext, hdfsEnvironment, hadoopPath(targetPath));
                 }
             }, fileSystemExecutor));
 
@@ -2028,9 +2029,9 @@ public class SemiTransactionalHiveMetastore
 
             Partition partition = partitionAndMore.getPartition();
             partitionsToInvalidate.add(partition);
-            Path targetPath = new Path(partition.getStorage().getLocation());
-            Path currentPath = partitionAndMore.getCurrentLocation();
-            cleanUpTasksForAbort.add(new DirectoryCleanUpTask(hdfsContext, targetPath, false));
+            Location targetPath = Location.of(partition.getStorage().getLocation());
+            Location currentPath = partitionAndMore.getCurrentLocation();
+            cleanUpTasksForAbort.add(new DirectoryCleanUpTask(hdfsContext, hadoopPath(targetPath), false));
 
             if (!targetPath.equals(currentPath)) {
                 // if staging directory is used we cherry-pick files to be moved
@@ -2499,21 +2500,21 @@ public class SemiTransactionalHiveMetastore
             AtomicBoolean cancelled,
             List<CompletableFuture<?>> fileRenameFutures,
             HdfsContext context,
-            Path currentPath,
-            Path targetPath,
+            Location currentPath,
+            Location targetPath,
             List<String> fileNames)
     {
         FileSystem fileSystem;
         try {
-            fileSystem = hdfsEnvironment.getFileSystem(context, currentPath);
+            fileSystem = hdfsEnvironment.getFileSystem(context, hadoopPath(currentPath));
         }
         catch (IOException e) {
             throw new TrinoException(HIVE_FILESYSTEM_ERROR, format("Error moving data files to final location. Error listing directory %s", currentPath), e);
         }
 
         for (String fileName : fileNames) {
-            Path source = new Path(currentPath, fileName);
-            Path target = new Path(targetPath, fileName);
+            Path source = hadoopPath(currentPath.appendPath(fileName));
+            Path target = hadoopPath(targetPath.appendPath(fileName));
             fileRenameFutures.add(CompletableFuture.runAsync(() -> {
                 if (cancelled.get()) {
                     return;
@@ -2830,7 +2831,7 @@ public class SemiTransactionalHiveMetastore
     {
         private final Table table;
         private final Optional<PrincipalPrivileges> principalPrivileges;
-        private final Optional<Path> currentLocation; // unpartitioned table only
+        private final Optional<Location> currentLocation; // unpartitioned table only
         private final Optional<List<String>> fileNames;
         private final boolean ignoreExisting;
         private final PartitionStatistics statistics;
@@ -2840,7 +2841,7 @@ public class SemiTransactionalHiveMetastore
         public TableAndMore(
                 Table table,
                 Optional<PrincipalPrivileges> principalPrivileges,
-                Optional<Path> currentLocation,
+                Optional<Location> currentLocation,
                 Optional<List<String>> fileNames,
                 boolean ignoreExisting,
                 PartitionStatistics statistics,
@@ -2876,7 +2877,7 @@ public class SemiTransactionalHiveMetastore
             return principalPrivileges.get();
         }
 
-        public Optional<Path> getCurrentLocation()
+        public Optional<Location> getCurrentLocation()
         {
             return currentLocation;
         }
@@ -2923,7 +2924,7 @@ public class SemiTransactionalHiveMetastore
         private final List<PartitionUpdateAndMergeResults> partitionMergeResults;
         private final List<Partition> partitions;
 
-        public TableAndMergeResults(Table table, Optional<PrincipalPrivileges> principalPrivileges, Optional<Path> currentLocation, List<PartitionUpdateAndMergeResults> partitionMergeResults, List<Partition> partitions)
+        public TableAndMergeResults(Table table, Optional<PrincipalPrivileges> principalPrivileges, Optional<Location> currentLocation, List<PartitionUpdateAndMergeResults> partitionMergeResults, List<Partition> partitions)
         {
             super(table, principalPrivileges, currentLocation, Optional.empty(), false, PartitionStatistics.empty(), PartitionStatistics.empty(), false); // retries are not supported for transactional tables
             this.partitionMergeResults = requireNonNull(partitionMergeResults, "partitionMergeResults is null");
@@ -2951,13 +2952,13 @@ public class SemiTransactionalHiveMetastore
     private static class PartitionAndMore
     {
         private final Partition partition;
-        private final Path currentLocation;
+        private final Location currentLocation;
         private final Optional<List<String>> fileNames;
         private final PartitionStatistics statistics;
         private final PartitionStatistics statisticsUpdate;
         private final boolean cleanExtraOutputFilesOnCommit;
 
-        public PartitionAndMore(Partition partition, Path currentLocation, Optional<List<String>> fileNames, PartitionStatistics statistics, PartitionStatistics statisticsUpdate, boolean cleanExtraOutputFilesOnCommit)
+        public PartitionAndMore(Partition partition, Location currentLocation, Optional<List<String>> fileNames, PartitionStatistics statistics, PartitionStatistics statisticsUpdate, boolean cleanExtraOutputFilesOnCommit)
         {
             this.partition = requireNonNull(partition, "partition is null");
             this.currentLocation = requireNonNull(currentLocation, "currentLocation is null");
@@ -2972,7 +2973,7 @@ public class SemiTransactionalHiveMetastore
             return partition;
         }
 
-        public Path getCurrentLocation()
+        public Location getCurrentLocation()
         {
             return currentLocation;
         }
@@ -3704,9 +3705,9 @@ public class SemiTransactionalHiveMetastore
         }
     }
 
-    public record PartitionUpdateInfo(List<String> partitionValues, Path currentLocation, List<String> fileNames, PartitionStatistics statisticsUpdate)
+    public record PartitionUpdateInfo(List<String> partitionValues, Location currentLocation, List<String> fileNames, PartitionStatistics statisticsUpdate)
     {
-        public PartitionUpdateInfo(List<String> partitionValues, Path currentLocation, List<String> fileNames, PartitionStatistics statisticsUpdate)
+        public PartitionUpdateInfo(List<String> partitionValues, Location currentLocation, List<String> fileNames, PartitionStatistics statisticsUpdate)
         {
             this.partitionValues = requireNonNull(partitionValues, "partitionValues is null");
             this.currentLocation = requireNonNull(currentLocation, "currentLocation is null");
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastore.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastore.java
index 41b8cec11b..c6c06608cb 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastore.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastore.java
@@ -69,12 +69,16 @@ import dev.failsafe.Failsafe;
 import dev.failsafe.RetryPolicy;
 import io.airlift.concurrent.MoreFutures;
 import io.airlift.log.Logger;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.filesystem.hdfs.HdfsFileSystemFactory;
 import io.trino.hdfs.DynamicHdfsConfiguration;
 import io.trino.hdfs.HdfsConfig;
 import io.trino.hdfs.HdfsConfiguration;
 import io.trino.hdfs.HdfsConfigurationInitializer;
 import io.trino.hdfs.HdfsContext;
 import io.trino.hdfs.HdfsEnvironment;
+import io.trino.hdfs.TrinoHdfsFileSystemStats;
 import io.trino.hdfs.authentication.NoHdfsAuthentication;
 import io.trino.plugin.hive.HiveColumnStatisticType;
 import io.trino.plugin.hive.HiveType;
@@ -185,6 +189,7 @@ public class GlueHiveMetastore
             .withMaxRetries(3)
             .build();
 
+    private final TrinoFileSystemFactory fileSystemFactory;
     private final HdfsEnvironment hdfsEnvironment;
     private final HdfsContext hdfsContext;
     private final AWSGlueAsync glueClient;
@@ -198,6 +203,7 @@ public class GlueHiveMetastore
 
     @Inject
     public GlueHiveMetastore(
+            TrinoFileSystemFactory fileSystemFactory,
             HdfsEnvironment hdfsEnvironment,
             GlueHiveMetastoreConfig glueConfig,
             @ForGlueHiveMetastore Executor partitionsReadExecutor,
@@ -206,6 +212,7 @@ public class GlueHiveMetastore
             @ForGlueHiveMetastore GlueMetastoreStats stats,
             @ForGlueHiveMetastore Predicate<com.amazonaws.services.glue.model.Table> tableFilter)
     {
+        this.fileSystemFactory = requireNonNull(fileSystemFactory, "fileSystemFactory is null");
         this.hdfsEnvironment = requireNonNull(hdfsEnvironment, "hdfsEnvironment is null");
         this.hdfsContext = new HdfsContext(ConnectorIdentity.ofUser(DEFAULT_METASTORE_USER));
         this.glueClient = requireNonNull(glueClient, "glueClient is null");
@@ -227,7 +234,9 @@ public class GlueHiveMetastore
         GlueMetastoreStats stats = new GlueMetastoreStats();
         GlueHiveMetastoreConfig glueConfig = new GlueHiveMetastoreConfig()
                 .setDefaultWarehouseDir(defaultWarehouseDir.toUri().toString());
+        TrinoFileSystemFactory fileSystemFactory = new HdfsFileSystemFactory(hdfsEnvironment, new TrinoHdfsFileSystemStats());
         return new GlueHiveMetastore(
+                fileSystemFactory,
                 hdfsEnvironment,
                 glueConfig,
                 directExecutor(),
@@ -555,7 +564,7 @@ public class GlueHiveMetastore
         }
 
         if (deleteData) {
-            location.ifPresent(path -> deleteDir(hdfsContext, hdfsEnvironment, new Path(path), true));
+            location.ifPresent(path -> deleteDir(hdfsContext, path));
         }
     }
 
@@ -621,7 +630,7 @@ public class GlueHiveMetastore
         Optional<String> location = table.getStorage().getOptionalLocation()
                 .filter(not(String::isEmpty));
         if (deleteData && isManagedTable(table) && location.isPresent()) {
-            deleteDir(hdfsContext, hdfsEnvironment, new Path(location.get()), true);
+            deleteDir(hdfsContext, location.get());
         }
     }
 
@@ -630,10 +639,10 @@ public class GlueHiveMetastore
         return table.getTableType().equals(MANAGED_TABLE.name());
     }
 
-    private static void deleteDir(HdfsContext context, HdfsEnvironment hdfsEnvironment, Path path, boolean recursive)
+    private void deleteDir(HdfsContext context, String path)
     {
         try {
-            hdfsEnvironment.getFileSystem(context, path).delete(path, recursive);
+            fileSystemFactory.create(context.getIdentity()).deleteDirectory(Location.of(path));
         }
         catch (Exception e) {
             // don't fail if unable to delete path
@@ -1082,7 +1091,7 @@ public class GlueHiveMetastore
 
         String partLocation = partition.getStorage().getLocation();
         if (deleteData && isManagedTable(table) && !isNullOrEmpty(partLocation)) {
-            deleteDir(hdfsContext, hdfsEnvironment, new Path(partLocation), true);
+            deleteDir(hdfsContext, partLocation);
         }
     }
 
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/RegisterPartitionProcedure.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/RegisterPartitionProcedure.java
index 7f7a737610..4a696a24e9 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/RegisterPartitionProcedure.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/RegisterPartitionProcedure.java
@@ -17,6 +17,7 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.inject.Inject;
 import com.google.inject.Provider;
+import io.trino.filesystem.Location;
 import io.trino.hdfs.HdfsContext;
 import io.trino.hdfs.HdfsEnvironment;
 import io.trino.plugin.hive.HiveConfig;
@@ -34,12 +35,12 @@ import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.connector.TableNotFoundException;
 import io.trino.spi.procedure.Procedure;
 import io.trino.spi.type.ArrayType;
-import org.apache.hadoop.fs.Path;
 
 import java.lang.invoke.MethodHandle;
 import java.util.List;
 import java.util.Optional;
 
+import static io.trino.filesystem.hdfs.HadoopPaths.hadoopPath;
 import static io.trino.plugin.base.util.Procedures.checkProcedureArgument;
 import static io.trino.plugin.hive.HiveMetadata.PRESTO_QUERY_ID_NAME;
 import static io.trino.plugin.hive.procedure.Procedures.checkIsPartitionedTable;
@@ -131,16 +132,16 @@ public class RegisterPartitionProcedure
             throw new TrinoException(ALREADY_EXISTS, format("Partition [%s] is already registered with location %s", partitionName, partition.get().getStorage().getLocation()));
         }
 
-        Path partitionLocation;
+        Location partitionLocation;
 
         if (location == null) {
-            partitionLocation = new Path(table.getStorage().getLocation(), makePartName(partitionColumns, partitionValues));
+            partitionLocation = Location.of(table.getStorage().getLocation()).appendPath(makePartName(partitionColumns, partitionValues));
         }
         else {
-            partitionLocation = new Path(location);
+            partitionLocation = Location.of(location);
         }
 
-        if (!HiveWriteUtils.pathExists(hdfsContext, hdfsEnvironment, partitionLocation)) {
+        if (!HiveWriteUtils.pathExists(hdfsContext, hdfsEnvironment, hadoopPath(partitionLocation))) {
             throw new TrinoException(INVALID_PROCEDURE_ARGUMENT, "Partition location does not exist: " + partitionLocation);
         }
 
@@ -157,7 +158,7 @@ public class RegisterPartitionProcedure
         metastore.commit();
     }
 
-    private static Partition buildPartitionObject(ConnectorSession session, Table table, List<String> partitionValues, Path location)
+    private static Partition buildPartitionObject(ConnectorSession session, Table table, List<String> partitionValues, Location location)
     {
         return Partition.builder()
                 .setDatabaseName(table.getDatabaseName())
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/SyncPartitionMetadataProcedure.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/SyncPartitionMetadataProcedure.java
index dd0e0983b1..f25b56bd99 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/SyncPartitionMetadataProcedure.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/SyncPartitionMetadataProcedure.java
@@ -19,6 +19,7 @@ import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import com.google.inject.Inject;
 import com.google.inject.Provider;
+import io.trino.filesystem.Location;
 import io.trino.hdfs.HdfsContext;
 import io.trino.hdfs.HdfsEnvironment;
 import io.trino.plugin.hive.PartitionStatistics;
@@ -240,7 +241,7 @@ public class SyncPartitionMetadataProcedure
                     table.getDatabaseName(),
                     table.getTableName(),
                     buildPartitionObject(session, table, name),
-                    new Path(table.getStorage().getLocation(), name),
+                    Location.of(table.getStorage().getLocation()).appendPath(name),
                     Optional.empty(), // no need for failed attempts cleanup
                     PartitionStatistics.empty(),
                     false);
diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/AbstractTestHive.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/AbstractTestHive.java
index 06cb66a91c..78bd3c4dd9 100644
--- a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/AbstractTestHive.java
+++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/AbstractTestHive.java
@@ -203,6 +203,7 @@ import static io.airlift.testing.Assertions.assertGreaterThanOrEqual;
 import static io.airlift.testing.Assertions.assertInstanceOf;
 import static io.airlift.testing.Assertions.assertLessThanOrEqual;
 import static io.airlift.units.DataSize.Unit.KILOBYTE;
+import static io.trino.filesystem.hdfs.HadoopPaths.hadoopPath;
 import static io.trino.parquet.reader.ParquetReader.PARQUET_CODEC_METRIC_PREFIX;
 import static io.trino.plugin.hive.AbstractTestHive.TransactionDeleteInsertTestTag.COMMIT;
 import static io.trino.plugin.hive.AbstractTestHive.TransactionDeleteInsertTestTag.ROLLBACK_AFTER_APPEND_PAGE;
@@ -6304,8 +6305,8 @@ public abstract class AbstractTestHive
                 throws IOException
         {
             for (PartitionUpdate partitionUpdate : partitionUpdates) {
-                if ("pk2=insert2".equals(partitionUpdate.getTargetPath().getName())) {
-                    path = new Path(partitionUpdate.getTargetPath(), partitionUpdate.getFileNames().get(0));
+                if ("pk2=insert2".equals(partitionUpdate.getTargetPath().fileName())) {
+                    path = new Path(hadoopPath(partitionUpdate.getTargetPath()), partitionUpdate.getFileNames().get(0));
                     break;
                 }
             }
diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/AbstractTestHiveLocal.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/AbstractTestHiveLocal.java
index f97a28461b..a3288f846e 100644
--- a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/AbstractTestHiveLocal.java
+++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/AbstractTestHiveLocal.java
@@ -18,6 +18,7 @@ import com.google.common.collect.ImmutableMap;
 import com.google.common.io.RecursiveDeleteOption;
 import com.google.common.reflect.ClassPath;
 import io.airlift.log.Logger;
+import io.trino.filesystem.Location;
 import io.trino.plugin.hive.metastore.Column;
 import io.trino.plugin.hive.metastore.Database;
 import io.trino.plugin.hive.metastore.HiveMetastore;
@@ -33,7 +34,6 @@ import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.predicate.TupleDomain;
 import io.trino.spi.security.PrincipalType;
 import io.trino.testing.MaterializedResult;
-import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.metastore.TableType;
 import org.testng.SkipException;
 import org.testng.annotations.AfterClass;
@@ -44,7 +44,6 @@ import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.UncheckedIOException;
-import java.net.URI;
 import java.nio.file.Files;
 import java.util.List;
 import java.util.Optional;
@@ -190,7 +189,7 @@ public abstract class AbstractTestHiveLocal
                             BUCKETING_V1,
                             3,
                             ImmutableList.of(new SortingColumn("name", SortingColumn.Order.ASCENDING)))),
-                    new Path(URI.create("file://" + externalLocation.toString())));
+                    Location.of("file://" + externalLocation.toString()));
 
             assertReadFailsWithMessageMatching(ORC, tableName, "Hive table is corrupt\\\\. File '.*/.*' is for bucket [0-2], but contains a row for bucket [0-2].");
             markTableAsCreatedBySpark(tableName, "orc");
@@ -227,7 +226,7 @@ public abstract class AbstractTestHiveLocal
         }
     }
 
-    private void createExternalTable(SchemaTableName schemaTableName, HiveStorageFormat hiveStorageFormat, List<Column> columns, List<Column> partitionColumns, Optional<HiveBucketProperty> bucketProperty, Path externalLocation)
+    private void createExternalTable(SchemaTableName schemaTableName, HiveStorageFormat hiveStorageFormat, List<Column> columns, List<Column> partitionColumns, Optional<HiveBucketProperty> bucketProperty, Location externalLocation)
     {
         try (Transaction transaction = newTransaction()) {
             ConnectorSession session = newSession();
diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveCreateExternalTable.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveCreateExternalTable.java
index 6ca1d28257..e897747666 100644
--- a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveCreateExternalTable.java
+++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveCreateExternalTable.java
@@ -51,13 +51,13 @@ public class TestHiveCreateExternalTable
             throws IOException
     {
         Path tempDir = createTempDirectory(null);
-        Path tableLocation = tempDir.resolve("data");
+        String tableLocation = tempDir.resolve("data").toUri().toASCIIString();
 
         @Language("SQL") String createTableSql = format("" +
                         "CREATE TABLE test_create_external " +
                         "WITH (external_location = '%s') AS " +
                         "SELECT * FROM tpch.tiny.nation",
-                tableLocation.toUri().toASCIIString());
+                tableLocation);
 
         assertUpdate(createTableSql, 25);
 
@@ -67,7 +67,7 @@ public class TestHiveCreateExternalTable
 
         MaterializedResult result = computeActual("SELECT DISTINCT regexp_replace(\\"$path\\", '/[^/]*$', '/') FROM test_create_external");
         String tablePath = (String) result.getOnlyValue();
-        assertThat(tablePath).startsWith(tableLocation.toFile().toURI().toString());
+        assertThat(tablePath).startsWith(tableLocation);
 
         assertUpdate("DROP TABLE test_create_external");
         deleteRecursively(tempDir, ALLOW_INSECURE);
diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveS3AndGlueMetastoreTest.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveS3AndGlueMetastoreTest.java
index e0f2b5d797..78600d318c 100644
--- a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveS3AndGlueMetastoreTest.java
+++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveS3AndGlueMetastoreTest.java
@@ -24,11 +24,9 @@ import org.testng.annotations.Test;
 
 import java.nio.file.Path;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 
-import static com.google.common.collect.ImmutableList.toImmutableList;
 import static io.trino.plugin.hive.metastore.glue.GlueHiveMetastore.createTestingGlueHiveMetastore;
 import static io.trino.spi.security.SelectedRole.Type.ROLE;
 import static io.trino.testing.TestingNames.randomNameSuffix;
@@ -37,6 +35,7 @@ import static java.util.Objects.requireNonNull;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
+// TODO Run this test also with Hive thrift metastore
 public class TestHiveS3AndGlueMetastoreTest
         extends BaseS3AndGlueMetastoreTest
 {
@@ -254,14 +253,55 @@ public class TestHiveS3AndGlueMetastoreTest
                 .hasStackTraceContaining("Fragment is not allowed in a file system location");
     }
 
+    @Test
+    public void testCreateTableWithDoubleSlash()
+    {
+        String schemaName = "test_create_table_with_double_slash_" + randomNameSuffix();
+        String schemaLocation = "s3://%s/%s/double_slash//test_schema".formatted(bucketName, schemaName);
+        String tableName = "test_create_table_with_double_slash_" + randomNameSuffix();
+        String schemaTableName = schemaName + "." + tableName;
+
+        // Previously, HiveLocationService replaced double slash with single slash
+        assertUpdate("CREATE SCHEMA " + schemaName + " WITH (location = '" + schemaLocation + "')");
+        String existingKey = "%s/double_slash/test_schema/%s".formatted(schemaName, tableName);
+        s3.putObject(bucketName, existingKey, "test content");
+
+        assertUpdate("CREATE TABLE " + schemaTableName + "(col_int int)");
+        assertUpdate("INSERT INTO " + schemaTableName + " VALUES 1", 1);
+        assertQuery("SELECT * FROM " + schemaTableName, "VALUES 1");
+        assertUpdate("DROP TABLE " + schemaTableName);
+        s3.deleteObject(bucketName, existingKey);
+    }
+
+    @Test
+    public void testCtasWithDoubleSlash()
+    {
+        String schemaName = "test_ctas_with_double_slash_" + randomNameSuffix();
+        String schemaLocation = "s3://%s/%s/double_slash//test_schema".formatted(bucketName, schemaName);
+        String tableName = "test_create_table_with_double_slash_" + randomNameSuffix();
+        String schemaTableName = schemaName + "." + tableName;
+
+        // Previously, HiveLocationService replaced double slash with single slash
+        assertUpdate("CREATE SCHEMA " + schemaName + " WITH (location = '" + schemaLocation + "')");
+        String existingKey = "%s/double_slash/test_schema/%s".formatted(schemaName, tableName);
+        s3.putObject(bucketName, existingKey, "test content");
+
+        assertUpdate("CREATE TABLE " + schemaTableName + " AS SELECT 1 AS col_int", 1);
+        assertQuery("SELECT * FROM " + schemaTableName, "VALUES 1");
+        assertUpdate("DROP TABLE " + schemaTableName);
+        s3.deleteObject(bucketName, existingKey);
+    }
+
     @Test
     public void testCreateSchemaWithIncorrectLocation()
     {
         String schemaName = "test_create_schema_with_incorrect_location_" + randomNameSuffix();
-        String schemaLocation = "s3://%s/%2$s/a#hash/%2$s".formatted(bucketName, schemaName);
+        String key = "%1$s/a#hash/%1$s";
+        String schemaLocation = "s3://%s/%s".formatted(bucketName, key);
         String tableName = "test_basic_operations_table_" + randomNameSuffix();
         String qualifiedTableName = schemaName + "." + tableName;
 
+        // TODO Disallow creating a schema with incorrect location
         assertUpdate("CREATE SCHEMA " + schemaName + " WITH (location = '" + schemaLocation + "')");
         assertThat(getSchemaLocation(schemaName)).isEqualTo(schemaLocation);
 
@@ -272,6 +312,8 @@ public class TestHiveS3AndGlueMetastoreTest
                 .hasMessageContaining("Fragment is not allowed in a file system location");
 
         assertUpdate("DROP SCHEMA " + schemaName);
+        // Delete S3 directory explicitly because the above DROP SCHEMA failed to delete it due to fragment in the location
+        s3.deleteObject(bucketName, key);
     }
 
     @Test
@@ -286,13 +328,4 @@ public class TestHiveS3AndGlueMetastoreTest
 
         assertUpdate("DROP SCHEMA \\"" + schemaName + "\\"");
     }
-
-    @Override
-    protected List<String> locationPatterns()
-    {
-        // TODO https://github.com/trinodb/trino/issues/17803 Fix correctness issue with double slash
-        return super.locationPatterns().stream()
-                .filter(location -> !location.contains("double_slash"))
-                .collect(toImmutableList());
-    }
 }
diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestPartitionUpdate.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestPartitionUpdate.java
index 69031a976a..52cb3aa1ea 100644
--- a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestPartitionUpdate.java
+++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestPartitionUpdate.java
@@ -15,8 +15,8 @@ package io.trino.plugin.hive;
 
 import com.google.common.collect.ImmutableList;
 import io.airlift.json.JsonCodec;
+import io.trino.filesystem.Location;
 import io.trino.plugin.hive.PartitionUpdate.UpdateMode;
-import org.apache.hadoop.fs.Path;
 import org.testng.annotations.Test;
 
 import static io.airlift.json.JsonCodec.jsonCodec;
@@ -43,8 +43,8 @@ public class TestPartitionUpdate
 
         assertEquals(actual.getName(), "test");
         assertEquals(actual.getUpdateMode(), UpdateMode.APPEND);
-        assertEquals(actual.getWritePath(), new Path("/writePath"));
-        assertEquals(actual.getTargetPath(), new Path("/targetPath"));
+        assertEquals(actual.getWritePath(), Location.of("/writePath"));
+        assertEquals(actual.getTargetPath(), Location.of("/targetPath"));
         assertEquals(actual.getFileNames(), ImmutableList.of("file1", "file3"));
         assertEquals(actual.getRowCount(), 123);
         assertEquals(actual.getInMemoryDataSizeInBytes(), 456);
diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/TestSemiTransactionalHiveMetastore.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/TestSemiTransactionalHiveMetastore.java
index df0a3f1b1a..836b011568 100644
--- a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/TestSemiTransactionalHiveMetastore.java
+++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/TestSemiTransactionalHiveMetastore.java
@@ -15,13 +15,13 @@ package io.trino.plugin.hive.metastore;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import io.trino.filesystem.Location;
 import io.trino.plugin.hive.HiveBucketProperty;
 import io.trino.plugin.hive.HiveMetastoreClosure;
 import io.trino.plugin.hive.HiveType;
 import io.trino.plugin.hive.PartitionStatistics;
 import io.trino.plugin.hive.acid.AcidTransaction;
 import io.trino.plugin.hive.fs.FileSystemDirectoryLister;
-import org.apache.hadoop.fs.Path;
 import org.testng.annotations.Test;
 
 import java.util.List;
@@ -53,7 +53,7 @@ public class TestSemiTransactionalHiveMetastore
             Optional.of("comment"));
     private static final Storage TABLE_STORAGE = new Storage(
             StorageFormat.create("serde", "input", "output"),
-            Optional.of("location"),
+            Optional.of("file://location/"),
             Optional.of(new HiveBucketProperty(ImmutableList.of("column"), BUCKETING_V1, 10, ImmutableList.of(new SortingColumn("column", SortingColumn.Order.ASCENDING)))),
             true,
             ImmutableMap.of("param", "value2"));
@@ -109,7 +109,7 @@ public class TestSemiTransactionalHiveMetastore
             IntStream.range(0, tablesToUpdate).forEach(i -> semiTransactionalHiveMetastore.finishChangingExistingTable(INSERT, SESSION,
                     "database",
                     "table_" + i,
-                    new Path("location"),
+                    Location.of("file://location/"),
                     ImmutableList.of(),
                     PartitionStatistics.empty(),
                     false));
diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestHiveGlueMetastore.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestHiveGlueMetastore.java
index 89da152872..c8e2264d22 100644
--- a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestHiveGlueMetastore.java
+++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestHiveGlueMetastore.java
@@ -30,6 +30,8 @@ import com.google.common.collect.ImmutableMap;
 import io.airlift.concurrent.BoundedExecutor;
 import io.airlift.log.Logger;
 import io.airlift.slice.Slice;
+import io.trino.filesystem.hdfs.HdfsFileSystemFactory;
+import io.trino.hdfs.TrinoHdfsFileSystemStats;
 import io.trino.plugin.hive.AbstractTestHiveLocal;
 import io.trino.plugin.hive.HiveBasicStatistics;
 import io.trino.plugin.hive.HiveMetastoreClosure;
@@ -232,6 +234,7 @@ public class TestHiveGlueMetastore
         Executor executor = new BoundedExecutor(this.executor, 10);
         GlueMetastoreStats stats = new GlueMetastoreStats();
         return new GlueHiveMetastore(
+                new HdfsFileSystemFactory(HDFS_ENVIRONMENT, new TrinoHdfsFileSystemStats()),
                 HDFS_ENVIRONMENT,
                 glueConfig,
                 executor,