[core] Fix compile error in DataFilePathFactory

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java

@@ -52,22 +52,21 @@ public DataFilePathFactory(
             String changelogFilePrefix) {
         this.parent = parent;
         this.uuid = UUID.randomUUID().toString();
-
         this.pathCount = new AtomicInteger(0);
         this.formatIdentifier = formatIdentifier;
         this.dataFilePrefix = dataFilePrefix;
         this.changelogFilePrefix = changelogFilePrefix;
     }
 
     public Path newPath() {
-        if (!StringUtils.isBlank(dataFilePrefix)) {
+        if (!StringUtils.isNullOrWhitespaceOnly(dataFilePrefix)) {
             return newPath(dataFilePrefix);
         }
         return newPath(DEFAULT_DATA_FILE_PREFIX);
     }
 
     public Path newChangelogPath() {
-        if (!StringUtils.isBlank(changelogFilePrefix)) {
+        if (!StringUtils.isNullOrWhitespaceOnly(changelogFilePrefix)) {
             return newPath(changelogFilePrefix);
         }
         return newPath(DEFAULT_CHANGELOG_FILE_PREFIX);

paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java

@@ -56,11 +56,9 @@
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
-import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
-import static org.apache.paimon.io.DataFilePathFactory.DEFAULT_CHANGELOG_FILE_PREFIX;
 import static org.apache.paimon.predicate.PredicateBuilder.containsFields;
 import static org.apache.paimon.predicate.PredicateBuilder.splitAnd;
 
@@ -299,28 +297,12 @@ public RecordReader<KeyValue> createNoMergeReader(
                         onlyFilterKey ? filtersForKeys : filtersForAll);
         List<ReaderSupplier<KeyValue>> suppliers = new ArrayList<>();
         for (DataFileMeta file : files) {
-            suppliers.add(
-                    () -> {
-                        // We need to check extraFiles to be compatible with Paimon 0.2.
-                        // See comments on DataFileMeta#extraFiles.
-                        String fileName = changelogFile(file).orElse(file.fileName());
-                        return readerFactory.createRecordReader(
-                                file.schemaId(), fileName, file.fileSize(), file.level());
-                    });
+            suppliers.add(() -> readerFactory.createRecordReader(file));
         }
 
         return projectOuter(ConcatRecordReader.create(suppliers));
     }
 
-    private Optional<String> changelogFile(DataFileMeta fileMeta) {
-        for (String file : fileMeta.extraFiles()) {
-            if (file.startsWith(DEFAULT_CHANGELOG_FILE_PREFIX)) {
-                return Optional.of(file);
-            }
-        }
-        return Optional.empty();
-    }
-
     private RecordReader<KeyValue> projectKey(RecordReader<KeyValue> reader) {
         if (keyProjectedFields == null) {
             return reader;