[pvfs] Fix file status and input stream for PVFS (#6397)

Author: timmyyao

paimon-vfs/paimon-vfs-hadoop/src/main/java/org/apache/paimon/vfs/hadoop/PaimonVirtualFileSystem.java

@@ -59,6 +59,7 @@ public class PaimonVirtualFileSystem extends FileSystem {
     private Configuration conf;
 
     private static final String USER_AGENT = "HadoopPVFS";
+    private static final long DEFAULT_BLOCK_SIZE = 128 * 1024 * 1024L;
 
     @Override
     public void initialize(URI uri, Configuration conf) throws IOException {
@@ -299,7 +300,7 @@ public boolean delete(Path f, boolean recursive) throws IOException {
     public FileStatus getFileStatus(Path f) throws IOException {
         VFSIdentifier vfsIdentifier = vfsOperations.getVFSIdentifier(getVirtualPath(f));
         if (vfsIdentifier instanceof VFSCatalogIdentifier) {
-            return new FileStatus(0, true, 1, 1, 0, new Path(this.uri));
+            return new FileStatus(0, true, 1, 0, 0, new Path(this.uri));
         } else if (vfsIdentifier instanceof VFSDatabaseIdentifier) {
             String databaseName = ((VFSDatabaseIdentifier) vfsIdentifier).databaseName();
             GetDatabaseResponse database = vfsOperations.getDatabase(databaseName);
@@ -318,7 +319,7 @@ public FileStatus getFileStatus(Path f) throws IOException {
     }
 
     private FileStatus convertDatabase(GetDatabaseResponse database) {
-        return new FileStatus(0, true, 1, 1, 0, new Path(new Path(this.uri), database.getName()));
+        return new FileStatus(0, true, 1, 0, 0, new Path(new Path(this.uri), database.getName()));
     }
 
     private FileStatus convertFileStatus(
@@ -338,11 +339,12 @@ private FileStatus convertFileStatus(
             childPath = "/" + childPath;
         }
         Path virtualPath = new Path(new Path(this.uri), databaseName + "/" + tableName + childPath);
+        long blockSize = fileStatus.isDir() ? 0 : DEFAULT_BLOCK_SIZE;
         return new FileStatus(
                 fileStatus.getLen(),
                 fileStatus.isDir(),
                 1,
-                1,
+                blockSize,
                 fileStatus.getModificationTime(),
                 virtualPath);
     }
@@ -375,7 +377,7 @@ private FileStatus[] convertDatabases(List<String> databases) {
         for (int i = 0; i < databases.size(); i++) {
             String database = databases.get(i);
             FileStatus fileStatus =
-                    new FileStatus(0, true, 1, 1, 0, new Path(new Path(this.uri), database));
+                    new FileStatus(0, true, 1, 0, 0, new Path(new Path(this.uri), database));
             fileStatuses[i] = fileStatus;
         }
         return fileStatuses;
@@ -387,7 +389,7 @@ private FileStatus[] convertTables(String database, List<String> tables) {
             String table = tables.get(i);
             FileStatus fileStatus =
                     new FileStatus(
-                            0, true, 1, 1, 0, new Path(new Path(this.uri), database + "/" + table));
+                            0, true, 1, 0, 0, new Path(new Path(this.uri), database + "/" + table));
             fileStatuses[i] = fileStatus;
         }
         return fileStatuses;

paimon-vfs/paimon-vfs-hadoop/src/main/java/org/apache/paimon/vfs/hadoop/VFSInputStream.java

@@ -77,4 +77,9 @@ public int read() throws IOException {
         }
         return (n == -1) ? -1 : oneByteBuf[0] & 0xff;
     }
+
+    @Override
+    public void close() throws IOException {
+        in.close();
+    }
 }

paimon-vfs/paimon-vfs-hadoop/src/test/java/org/apache/paimon/vfs/hadoop/VirtualFileSystemTest.java

@@ -137,6 +137,7 @@ public void testMkdir() throws Exception {
         FileStatus fileStatus = vfs.getFileStatus(vfsPath);
         Assert.assertEquals(vfsPath.toString(), fileStatus.getPath().toString());
         Assert.assertTrue(fileStatus.isDirectory());
+        Assert.assertEquals(0, fileStatus.getBlockSize());
 
         // Mkdir in non-existing table
         tableName = "object_table2";
@@ -195,6 +196,7 @@ public void testCreate() throws Exception {
         Assert.assertEquals(vfsPath.toString(), fileStatus.getPath().toString());
         Assert.assertTrue(fileStatus.isFile());
         Assert.assertEquals(5, fileStatus.getLen());
+        Assert.assertEquals(128 * 1024 * 1024L, fileStatus.getBlockSize());
 
         FSDataInputStream in = vfs.open(vfsPath);
         byte[] buffer = new byte[5];