[pvfs] Set working directory to the catalog root in PaimonVirtualFileSystem

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

@@ -67,11 +67,13 @@ public void initialize(URI uri, Configuration conf) throws IOException {
         this.conf = conf;
         super.initialize(uri, conf);
 
-        this.workingDirectory = new Path(uri);
         if (uri.getAuthority() == null || uri.getAuthority().isEmpty()) {
             throw new IllegalArgumentException("URI authority is empty: " + uri);
         }
         this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority() + "/");
+        // Hadoop passes the full user URI here, path component included, so the working
+        // directory must be derived from the normalized catalog root.
+        this.workingDirectory = new Path(this.uri);
 
         initVFSOperations();
     }

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

@@ -473,6 +473,30 @@ public void testVisitNormalTable() throws Exception {
                 new Path(vfsPath, "schema").toString(), fileStatuses[0].getPath().toString());
     }
 
+    @Test
+    public void testWorkingDirectoryIsCatalogRoot() throws Exception {
+        String databaseName = "test_db";
+        String tableName = "object_table";
+        createObjectTable(databaseName, tableName);
+
+        // Hadoop hands initialize() the full user URI, path component included
+        Path filePath = new Path(vfsRoot, databaseName + "/" + tableName + "/a.csv");
+        try (FileSystem fs = new PaimonVirtualFileSystem()) {
+            fs.initialize(filePath.toUri(), vfs.getConf());
+            assertThat(fs.getWorkingDirectory()).isEqualTo(vfsRoot);
+
+            // A relative path resolves against the catalog root, so it lands in the table it
+            // names rather than in the table that happened to open this file system
+            String relative = databaseName + "/" + tableName + "2/b.csv";
+            FSDataOutputStream out = fs.create(new Path(relative));
+            out.write("hello".getBytes());
+            out.close();
+
+            assertThat(fs.exists(new Path(vfsRoot, relative))).isTrue();
+            assertThat(fs.exists(filePath)).isFalse();
+        }
+    }
+
     @Test
     public void testTrash() throws Exception {
         String databaseName = "test_db";