Add "hive.fs.new-file-inherit-permissions".

fix bug.

Add test.

lib/trino-hdfs/src/main/java/io/trino/filesystem/hdfs/HdfsFileSystem.java

@@ -25,6 +25,7 @@
 import io.trino.hdfs.HdfsContext;
 import io.trino.hdfs.HdfsEnvironment;
 import io.trino.hdfs.TrinoHdfsFileSystemStats;
+import io.trino.spi.TrinoException;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -49,6 +50,7 @@
 import static io.trino.filesystem.hdfs.HadoopPaths.hadoopPath;
 import static io.trino.filesystem.hdfs.HdfsFileIterator.listedLocation;
 import static io.trino.hdfs.FileSystemUtils.getRawFileSystem;
+import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 import static java.util.UUID.randomUUID;
 import static java.util.stream.Collectors.groupingBy;
@@ -296,8 +298,16 @@ public void createDirectory(Location location)
             if (!hierarchical(fileSystem, location)) {
                 return null;
             }
-            Optional<FsPermission> permission = environment.getNewDirectoryPermissions();
-            try (TimeStat.BlockTimer _ = stats.getCreateDirectoryCalls().time()) {
+
+            Optional<FsPermission> permission = Optional.empty();
+            if (environment.getNewDirectoryPermissions().isPresent()) {
+                permission = environment.getNewDirectoryPermissions();
+            }
+            else if (environment.isNewFileInheritPermissions()){
+                permission = Optional.of(getExistingParentDirectoryPermission(fileSystem, directory));
+            }
+
+           try (TimeStat.BlockTimer _ = stats.getCreateDirectoryCalls().time()) {
                 if (!fileSystem.mkdirs(directory, permission.orElse(null))) {
                     throw new IOException("mkdirs failed");
                 }
@@ -417,6 +427,9 @@ public Optional<Location> createTemporaryDirectory(Location targetLocation, Stri
                 if (permission.isPresent()) {
                     fileSystem.setPermission(temporaryPath, permission.get());
                 }
+                else if (environment.isNewFileInheritPermissions()) {
+                    inheritDirectoryPermission(fileSystem, temporaryPath, targetPath);
+                }
 
                 return Optional.of(temporaryLocation);
             }
@@ -455,6 +468,36 @@ private boolean hierarchical(FileSystem fileSystem, Location rootLocation)
         }
     }
 
+    public static FsPermission getExistingParentDirectoryPermission(FileSystem fileSystem, Path path)
+            throws IOException
+    {
+        try {
+            // find the parent-directory where it exists
+            Path checkPath = path;
+            while (!fileSystem.exists(checkPath)){
+                checkPath = checkPath.getParent();
+            }
+
+            // return the parent-directory permission
+            return fileSystem.getFileStatus(checkPath).getPermission();
+        }
+        catch (IOException e) {
+            throw new IOException("Failed to get permission on exist-parent-directory for %s: %s".formatted(path.toString(), e.getMessage()), e);
+        }
+    }
+
+    private static void inheritDirectoryPermission(FileSystem fileSystem, Path path, Path targetPath)
+            throws IOException
+    {
+        try {
+            FsPermission fsPermission = getExistingParentDirectoryPermission(fileSystem, targetPath);
+            fileSystem.setPermission(path, fsPermission);
+        }
+        catch (IOException e) {
+            throw new IOException("Failed to set permission on %s based on %s: %s".formatted(path.toString(), targetPath.toString(), e.getMessage()), e);
+        }
+    }
+
     static <T extends Throwable> T withCause(T throwable, Throwable cause)
     {
         throwable.initCause(cause);

lib/trino-hdfs/src/main/java/io/trino/hdfs/HdfsConfig.java

@@ -41,6 +41,7 @@ public class HdfsConfig
     private List<File> resourceConfigFiles = ImmutableList.of();
     private String newDirectoryPermissions = "0777";
     private boolean newFileInheritOwnership;
+    private boolean newFileInheritPermissions = false;
     private boolean verifyChecksum = true;
     private Duration ipcPingInterval = new Duration(10, TimeUnit.SECONDS);
     private Duration dfsTimeout = new Duration(60, TimeUnit.SECONDS);
@@ -103,6 +104,19 @@ public HdfsConfig setNewFileInheritOwnership(boolean newFileInheritOwnership)
         return this;
     }
 
+    public boolean isNewFileInheritPermissions()
+    {
+        return newFileInheritPermissions;
+    }
+
+    @Config("hive.fs.new-file-inherit-permissions")
+    @ConfigDescription("Flag to determine if new files inherit the permissions information from the directory.")
+    public HdfsConfig setNewFileInheritPermissions(boolean newFileInheritPermissions)
+    {
+        this.newFileInheritPermissions = newFileInheritPermissions;
+        return this;
+    }
+
     public boolean isVerifyChecksum()
     {
         return verifyChecksum;

lib/trino-hdfs/src/main/java/io/trino/hdfs/HdfsEnvironment.java

@@ -53,6 +53,7 @@ public class HdfsEnvironment
     private final HdfsAuthentication hdfsAuthentication;
     private final Optional<FsPermission> newDirectoryPermissions;
     private final boolean newFileInheritOwnership;
+    private final boolean newFileInheritPermissions;
     private final boolean verifyChecksum;
     private final Optional<GcsStorageFactory> gcsStorageFactory;
 
@@ -73,6 +74,7 @@ public HdfsEnvironment(
         this.openTelemetry = requireNonNull(openTelemetry, "openTelemetry is null");
         this.hdfsConfiguration = requireNonNull(hdfsConfiguration, "hdfsConfiguration is null");
         this.newFileInheritOwnership = config.isNewFileInheritOwnership();
+        this.newFileInheritPermissions = config.isNewFileInheritPermissions();
         this.verifyChecksum = config.isVerifyChecksum();
         this.hdfsAuthentication = requireNonNull(hdfsAuthentication, "hdfsAuthentication is null");
         this.newDirectoryPermissions = config.getNewDirectoryFsPermissions();
@@ -127,6 +129,11 @@ public boolean isNewFileInheritOwnership()
         return newFileInheritOwnership;
     }
 
+    public boolean isNewFileInheritPermissions()
+    {
+        return newFileInheritPermissions;
+    }
+
     public <T> T doAs(ConnectorIdentity identity, ExceptionAction<T> action)
             throws IOException
     {

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

@@ -28,6 +28,7 @@
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
@@ -144,4 +145,37 @@ private void assertCreateDirectoryPermission(TrinoFileSystem fileSystem, HdfsEnv
         FileStatus status = hdfsEnvironment.getFileSystem(hdfsContext, path).getFileStatus(path);
         assertThat(status.getPermission().toOctal()).isEqualTo(permission);
     }
+
+    @Test
+    public void testInheritNewDirectoryPermissions()
+            throws Exception
+    {
+        HdfsConfig configWithInherit = new HdfsConfig()
+                .setNewDirectoryPermissions(HdfsConfig.SKIP_DIR_PERMISSIONS)
+                .setNewFileInheritPermissions(true);
+        HdfsEnvironment hdfsEnvironment = new HdfsEnvironment(hdfsConfiguration, configWithInherit, new NoHdfsAuthentication());
+        TrinoFileSystem fileSystem = new HdfsFileSystem(hdfsEnvironment, hdfsContext, new TrinoHdfsFileSystemStats());
+
+        short testPermission = (short) 740;
+
+        // create parent directory
+        Location parentLocation = getRootLocation().appendPath("test");
+        fileSystem.createDirectory(parentLocation);
+        Path path = new Path(parentLocation.toString());
+        hdfsEnvironment.getFileSystem(hdfsContext, path).setPermission(path, new FsPermission(testPermission));
+
+        // test single directory
+        Location inheritLocation = parentLocation.appendPath("inherit");
+        fileSystem.createDirectory(inheritLocation);
+        Path inheritPath = new Path(inheritLocation.toString());
+        FileStatus inheritStatus = hdfsEnvironment.getFileSystem(hdfsContext, inheritPath).getFileStatus(inheritPath);
+        assertThat(inheritStatus.getPermission().toOctal()).isEqualTo(testPermission);
+
+        // test multi directories
+        Location inheritMultiLocation = parentLocation.appendPath("partition/inherit");
+        fileSystem.createDirectory(inheritMultiLocation);
+        Path inheritMultiPath = new Path(inheritMultiLocation.toString());
+        FileStatus inheritMultiStatus = hdfsEnvironment.getFileSystem(hdfsContext, inheritMultiPath).getFileStatus(inheritMultiPath);
+        assertThat(inheritMultiStatus .getPermission().toOctal()).isEqualTo(testPermission);
+    }
 }

lib/trino-hdfs/src/test/java/io/trino/hdfs/TestHdfsConfig.java

@@ -40,6 +40,7 @@ public void testDefaults()
                 .setResourceConfigFiles(ImmutableList.of())
                 .setNewDirectoryPermissions("0777")
                 .setNewFileInheritOwnership(false)
+                .setNewFileInheritPermissions(false)
                 .setVerifyChecksum(true)
                 .setIpcPingInterval(new Duration(10, TimeUnit.SECONDS))
                 .setDfsTimeout(new Duration(60, TimeUnit.SECONDS))
@@ -64,6 +65,7 @@ public void testExplicitPropertyMappings()
                 .put("hive.config.resources", resource1.toString() + "," + resource2.toString())
                 .put("hive.fs.new-directory-permissions", "0700")
                 .put("hive.fs.new-file-inherit-ownership", "true")
+                .put("hive.fs.new-file-inherit-permissions", "true")
                 .put("hive.dfs.verify-checksum", "false")
                 .put("hive.dfs.ipc-ping-interval", "34s")
                 .put("hive.dfs-timeout", "33s")
@@ -81,6 +83,7 @@ public void testExplicitPropertyMappings()
                 .setResourceConfigFiles(ImmutableList.of(resource1.toFile().getAbsolutePath(), resource2.toFile().getAbsolutePath()))
                 .setNewDirectoryPermissions("0700")
                 .setNewFileInheritOwnership(true)
+                .setNewFileInheritPermissions(true)
                 .setVerifyChecksum(false)
                 .setIpcPingInterval(new Duration(34, TimeUnit.SECONDS))
                 .setDfsTimeout(new Duration(33, TimeUnit.SECONDS))