test: add testListPartitionNamesIsolationAcrossTables

Signed-off-by: pufan <[email protected]>

Author: 2k124

fe/fe-core/src/test/java/com/starrocks/connector/paimon/PaimonMetadataTest.java

@@ -71,6 +71,7 @@
 import mockit.MockUp;
 import mockit.Mocked;
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CachingCatalog;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.catalog.CatalogFactory;
@@ -83,6 +84,7 @@
 import org.apache.paimon.data.Timestamp;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.partition.Partition;
 import org.apache.paimon.predicate.Predicate;
@@ -130,6 +132,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -1013,4 +1016,72 @@ public void testGetTableVersionRange() throws Exception {
         catalog.dropDatabase("test_db", true, true);
         Files.delete(tmpDir);
     }
+
+    @Test
+    public void testListPartitionNamesIsolationAcrossTables(@Mocked FileStoreTable mockPaimonTable1,
+                                                            @Mocked FileStoreTable mockPaimonTable2)
+            throws Catalog.TableNotExistException {
+
+        Options options = new Options();
+        options.set(CatalogOptions.CACHE_ENABLED, true);
+        Catalog cachingCatalog = CachingCatalog.tryToCreate(paimonNativeCatalog, options);
+        PaimonMetadata newMetadata = new PaimonMetadata("test_catalog", new HdfsEnvironment(), cachingCatalog,
+                new ConnectorProperties(ConnectorType.PAIMON));
+
+        Identifier tblIdentifier1 = new Identifier("db1", "tbl1");
+        List<String> partitionKeys1 = Lists.newArrayList("year", "month");
+        Identifier tblIdentifier2 = new Identifier("db2", "tbl2");
+        List<String> partitionKeys2 = Lists.newArrayList("year", "month");
+
+        RowType tblRowType1 = RowType.of(new DataType[] {new IntType(true), new IntType(true)}, new String[] {"year", "month"});
+        RowType tblRowType2 = RowType.of(new DataType[] {new IntType(true), new IntType(true)}, new String[] {"year", "month"});
+
+        org.apache.paimon.partition.Partition db1PaimonPartition1 =
+                new org.apache.paimon.partition.Partition(new LinkedHashMap<String, String>() {{
+                    put("year", "2020");
+                    put("month", "1");
+                }}, 100L, 2048L, 2L, System.currentTimeMillis(), false);
+        org.apache.paimon.partition.Partition db2PaimonPartition1 =
+                new org.apache.paimon.partition.Partition(new LinkedHashMap<String, String>() {{
+                    put("year", "2021");
+                    put("month", "1");
+                }}, 100L, 2048L, 2L, System.currentTimeMillis(), false);
+        org.apache.paimon.partition.Partition db2PaimonPartition2 =
+                new org.apache.paimon.partition.Partition(new LinkedHashMap<String, String>() {{
+                    put("year", "2022");
+                    put("month", "1");
+                }}, 100L, 2048L, 2L, System.currentTimeMillis(), false);
+
+        new Expectations() {
+            {
+                // Table 1
+                paimonNativeCatalog.getTable(tblIdentifier1);
+                result = mockPaimonTable1;
+                paimonNativeCatalog.listPartitions(tblIdentifier1);
+                result = Lists.newArrayList(db1PaimonPartition1);
+                mockPaimonTable1.partitionKeys();
+                result = partitionKeys1;
+                mockPaimonTable1.rowType();
+                result = tblRowType1;
+
+                // Table 2
+                paimonNativeCatalog.getTable(tblIdentifier2);
+                result = mockPaimonTable2;
+                paimonNativeCatalog.listPartitions(tblIdentifier2);
+                result = Lists.newArrayList(db2PaimonPartition1, db2PaimonPartition2);
+                mockPaimonTable2.partitionKeys();
+                result = partitionKeys2;
+                mockPaimonTable2.rowType();
+                result = tblRowType2;
+
+            }
+        };
+
+        List<String> result1 = newMetadata.listPartitionNames("db1", "tbl1", null);
+        List<String> result2 = newMetadata.listPartitionNames("db2", "tbl2", null);
+
+        // Before fix, if the key does not contain "db" and "table", it will return 2. After fix, it will return 3.
+        assertEquals(2, result2.size());
+
+    }
 }