[core] Fix partition timestamp pattern replacement for prefix variable names

paimon-core/src/main/java/org/apache/paimon/partition/PartitionTimeExtractor.java

@@ -96,16 +96,48 @@ public LocalDateTime extract(List<String> partitionKeys, List<?> partitionValues
         if (pattern == null) {
             timestampString = partitionValues.get(0).toString();
         } else {
-            timestampString = pattern;
-            for (int i = 0; i < partitionKeys.size(); i++) {
-                timestampString =
-                        timestampString.replaceAll(
-                                "\\$" + partitionKeys.get(i), partitionValues.get(i).toString());
-            }
+            timestampString = replacePattern(pattern, partitionKeys, partitionValues);
         }
         return toLocalDateTime(timestampString, this.formatter);
     }
 
+    private static String replacePattern(
+            String pattern, List<String> partitionKeys, List<?> partitionValues) {
+        // Prefer longer keys so that a short key does not accidentally match a prefix of a
+        // longer key (e.g. $dt should not match inside $dt1).
+        List<Integer> indices = new ArrayList<>();
+        for (int i = 0; i < partitionKeys.size(); i++) {
+            indices.add(i);
+        }
+        indices.sort(
+                (i, j) ->
+                        Integer.compare(
+                                partitionKeys.get(j).length(), partitionKeys.get(i).length()));
+
+        StringBuilder builder = new StringBuilder(pattern.length());
+        for (int i = 0; i < pattern.length(); ) {
+            char c = pattern.charAt(i);
+            if (c == '$' && i + 1 < pattern.length()) {
+                String matchedValue = null;
+                for (int idx : indices) {
+                    String key = partitionKeys.get(idx);
+                    if (pattern.regionMatches(i + 1, key, 0, key.length())) {
+                        matchedValue = partitionValues.get(idx).toString();
+                        i += 1 + key.length();
+                        break;
+                    }
+                }
+                if (matchedValue != null) {
+                    builder.append(matchedValue);
+                    continue;
+                }
+            }
+            builder.append(c);
+            i++;
+        }
+        return builder.toString();
+    }
+
     private static LocalDateTime toLocalDateTime(
             String timestampString, @Nullable String formatterPattern) {
 

paimon-core/src/test/java/org/apache/paimon/partition/PartitionTimeExtractorTest.java

@@ -92,6 +92,28 @@ public void testFormatter() {
                 .isEqualTo(LocalDateTime.parse("2023-01-01T00:00:00"));
     }
 
+    @Test
+    public void testPatternWithPrefixVariableNames() {
+        PartitionTimeExtractor extractor =
+                new PartitionTimeExtractor("$dt1-$dt-$dt2 00:00:00", null);
+        assertThat(
+                        extractor.extract(
+                                Arrays.asList("dt1", "dt", "dt2"),
+                                Arrays.asList("2023", "01", "02")))
+                .isEqualTo(LocalDateTime.parse("2023-01-02T00:00:00"));
+
+        extractor = new PartitionTimeExtractor("$t $t2:$t3:$t4", null);
+        assertThat(
+                        extractor.extract(
+                                Arrays.asList("t", "t4", "t2", "t3"),
+                                Arrays.asList("2023-01-01", "03", "01", "02")))
+                .isEqualTo(LocalDateTime.parse("2023-01-01T01:02:03"));
+
+        extractor = new PartitionTimeExtractor("$dt1$dt", "MMyyyydd");
+        assertThat(extractor.extract(Arrays.asList("dt", "dt1"), Arrays.asList("01", "022023")))
+                .isEqualTo(LocalDateTime.parse("2023-02-01T00:00:00"));
+    }
+
     @Test
     public void testExtractNonDateFormattedPartition() {
         PartitionTimeExtractor extractor = new PartitionTimeExtractor("$ds", "yyyyMMdd");