Remove Hadoop usage from SemiTransactionalHiveMetastore

lib/trino-hdfs/src/test/java/io/trino/filesystem/hdfs/TestHdfsFileSystemHdfs.java

@@ -31,6 +31,7 @@
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -42,6 +43,7 @@ public class TestHdfsFileSystemHdfs
         extends AbstractTestTrinoFileSystem
 {
     private Hadoop hadoop;
+    private HdfsConfiguration hdfsConfiguration;
     private HdfsEnvironment hdfsEnvironment;
     private HdfsContext hdfsContext;
     private TrinoFileSystem fileSystem;
@@ -53,7 +55,7 @@ void beforeAll()
         hadoop.start();
 
         HdfsConfig hdfsConfig = new HdfsConfig();
-        HdfsConfiguration hdfsConfiguration = new DynamicHdfsConfiguration(new HdfsConfigurationInitializer(hdfsConfig), emptySet());
+        hdfsConfiguration = new DynamicHdfsConfiguration(new HdfsConfigurationInitializer(hdfsConfig), emptySet());
         hdfsEnvironment = new HdfsEnvironment(hdfsConfiguration, hdfsConfig, new NoHdfsAuthentication());
         hdfsContext = new HdfsContext(ConnectorIdentity.ofUser("test"));
 
@@ -107,4 +109,33 @@ protected void verifyFileSystemIsEmpty()
             throw new UncheckedIOException(e);
         }
     }
+
+    @Test
+    void testCreateDirectoryPermission()
+            throws IOException
+    {
+        assertCreateDirectoryPermission(fileSystem, hdfsEnvironment, (short) 777);
+    }
+
+    @Test
+    void testCreateDirectoryPermissionWithSkip()
+            throws IOException
+    {
+        HdfsConfig configWithSkip = new HdfsConfig()
+                .setNewDirectoryPermissions(HdfsConfig.SKIP_DIR_PERMISSIONS);
+        HdfsEnvironment hdfsEnvironment = new HdfsEnvironment(hdfsConfiguration, configWithSkip, new NoHdfsAuthentication());
+        TrinoFileSystem fileSystem = new HdfsFileSystem(hdfsEnvironment, hdfsContext, new TrinoHdfsFileSystemStats());
+
+        assertCreateDirectoryPermission(fileSystem, hdfsEnvironment, (short) 755);
+    }
+
+    private void assertCreateDirectoryPermission(TrinoFileSystem fileSystem, HdfsEnvironment hdfsEnvironment, short permission)
+            throws IOException
+    {
+        Location location = getRootLocation().appendPath("test");
+        fileSystem.createDirectory(location);
+        Path path = new Path(location.toString());
+        FileStatus status = hdfsEnvironment.getFileSystem(hdfsContext, path).getFileStatus(path);
+        assertThat(status.getPermission().toOctal()).isEqualTo(permission);
+    }
 }

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java

@@ -32,7 +32,6 @@
 import io.trino.filesystem.Location;
 import io.trino.filesystem.TrinoFileSystem;
 import io.trino.filesystem.TrinoFileSystemFactory;
-import io.trino.hdfs.HdfsContext;
 import io.trino.hdfs.HdfsEnvironment;
 import io.trino.plugin.base.CatalogName;
 import io.trino.plugin.base.projection.ApplyProjectionUtil;
@@ -127,7 +126,6 @@
 import io.trino.spi.type.VarcharType;
 import org.apache.avro.Schema;
 import org.apache.avro.SchemaParseException;
-import org.apache.hadoop.fs.Path;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
@@ -1833,7 +1831,7 @@ public Optional<ConnectorOutputMetadata> finishCreateTable(ConnectorSession sess
             tableStatistics = new PartitionStatistics(createEmptyStatistics(), ImmutableMap.of());
         }
 
-        Optional<Path> writePath = Optional.of(new Path(writeInfo.writePath().toString()));
+        Optional<Location> writePath = Optional.of(writeInfo.writePath());
         if (handle.getPartitionedBy().isEmpty()) {
             List<String> fileNames;
             if (partitionUpdates.isEmpty()) {
@@ -2208,7 +2206,7 @@ private Table finishChangingTable(AcidOperation acidOperation, String changeDesc
                             session,
                             table,
                             principalPrivileges,
-                            Optional.of(new Path(partitionUpdate.getWritePath().toString())),
+                            Optional.of(partitionUpdate.getWritePath()),
                             Optional.of(partitionUpdate.getFileNames()),
                             false,
                             partitionStatistics,
@@ -2268,8 +2266,8 @@ else if (partitionUpdate.getUpdateMode() == NEW || partitionUpdate.getUpdateMode
                     if (handle.getLocationHandle().getWriteMode() == DIRECT_TO_TARGET_EXISTING_DIRECTORY) {
                         removeNonCurrentQueryFiles(session, partitionUpdate.getTargetPath());
                         if (handle.isRetriesEnabled()) {
-                            HdfsContext hdfsContext = new HdfsContext(session);
-                            cleanExtraOutputFiles(hdfsEnvironment, hdfsContext, session.getQueryId(), partitionUpdate.getTargetPath(), ImmutableSet.copyOf(partitionUpdate.getFileNames()));
+                            TrinoFileSystem fileSystem = fileSystemFactory.create(session);
+                            cleanExtraOutputFiles(fileSystem, session.getQueryId(), partitionUpdate.getTargetPath(), ImmutableSet.copyOf(partitionUpdate.getFileNames()));
                         }
                     }
                     else {
@@ -2303,7 +2301,6 @@ private void removeNonCurrentQueryFiles(ConnectorSession session, Location parti
         String queryId = session.getQueryId();
         TrinoFileSystem fileSystem = fileSystemFactory.create(session);
         try {
-            // use TrinoFileSystem instead of Hadoop file system
             FileIterator iterator = fileSystem.listFiles(partitionLocation);
             while (iterator.hasNext()) {
                 Location location = iterator.next().location();

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadataFactory.java

@@ -238,7 +238,7 @@ public TransactionalMetadata create(ConnectorIdentity identity, boolean autoComm
 
         DirectoryLister directoryLister = transactionScopeCachingDirectoryListerFactory.get(this.directoryLister);
         SemiTransactionalHiveMetastore metastore = new SemiTransactionalHiveMetastore(
-                hdfsEnvironment,
+                fileSystemFactory,
                 hiveMetastoreClosure,
                 fileSystemExecutor,
                 dropExecutor,