[core]save the runtime memory while reading all the partitions or buckets

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntryCache.java

@@ -112,8 +112,10 @@ protected ManifestEntrySegments createSegments(Path path, @Nullable Long fileSiz
     }
 
     @Override
-    protected List<ManifestEntry> readFromSegments(
-            ManifestEntrySegments manifestSegments, Filters<ManifestEntry> filters)
+    protected <R> List<R> readFromSegments(
+            ManifestEntrySegments manifestSegments,
+            Filters<ManifestEntry> filters,
+            Function<ManifestEntry, R> convertor)
             throws IOException {
         PartitionPredicate partitionFilter = null;
         BucketFilter bucketFilter = null;
@@ -160,12 +162,16 @@ protected List<ManifestEntry> readFromSegments(
         }
 
         // read manifest entries from segments with per record filter
-        List<ManifestEntry> result = new ArrayList<>();
+        List<R> result = new ArrayList<>();
         InternalRowSerializer formatSerializer = this.formatSerializer.get();
         for (Segments subSegments : segmentsList) {
             result.addAll(
                     SimpleObjectsCache.readFromSegments(
-                            formatSerializer, projectedSerializer, subSegments, filters));
+                            formatSerializer,
+                            projectedSerializer,
+                            subSegments,
+                            filters,
+                            convertor));
         }
         return result;
     }

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

@@ -367,7 +367,10 @@ public List<PartitionEntry> readPartitionEntries() {
         List<ManifestFileMeta> manifests = readManifests().filteredManifests;
         Map<BinaryRow, PartitionEntry> partitions = new ConcurrentHashMap<>();
         Consumer<ManifestFileMeta> processor =
-                m -> PartitionEntry.merge(PartitionEntry.merge(readManifest(m)), partitions);
+                m ->
+                        PartitionEntry.merge(
+                                readManifest(m, PartitionEntry::fromManifestEntry, null, null),
+                                partitions);
         randomlyOnlyExecute(getExecutorService(parallelism), processor, manifests);
         return partitions.values().stream()
                 .filter(p -> p.fileCount() > 0)
@@ -379,7 +382,10 @@ public List<BucketEntry> readBucketEntries() {
         List<ManifestFileMeta> manifests = readManifests().filteredManifests;
         Map<Pair<BinaryRow, Integer>, BucketEntry> buckets = new ConcurrentHashMap<>();
         Consumer<ManifestFileMeta> processor =
-                m -> BucketEntry.merge(BucketEntry.merge(readManifest(m)), buckets);
+                m ->
+                        BucketEntry.merge(
+                                readManifest(m, BucketEntry::fromManifestEntry, null, null),
+                                buckets);
         randomlyOnlyExecute(getExecutorService(parallelism), processor, manifests);
         return buckets.values().stream()
                 .filter(p -> p.fileCount() > 0)

paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java

@@ -28,7 +28,6 @@
 import javax.annotation.concurrent.ThreadSafe;
 
 import java.io.IOException;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Function;
 
@@ -77,7 +76,7 @@ public <R> List<R> read(
             if (cacheMetrics != null) {
                 cacheMetrics.increaseHitObject();
             }
-            return convert(readFromSegments(segments, filters), convertor);
+            return readFromSegments(segments, filters, convertor);
         } else {
             if (cacheMetrics != null) {
                 cacheMetrics.increaseMissedObject();
@@ -88,7 +87,7 @@ public <R> List<R> read(
             if (fileSize <= cache.maxElementSize()) {
                 segments = createSegments(key, fileSize);
                 cache.put(key, segments);
-                return convert(readFromSegments(segments, filters), convertor);
+                return readFromSegments(segments, filters, convertor);
             } else {
                 return readFromIterator(
                         reader.apply(key, fileSize),
@@ -100,15 +99,8 @@ public <R> List<R> read(
         }
     }
 
-    private <R> List<R> convert(List<V> values, Function<V, R> convertor) {
-        List<R> result = new ArrayList<>(values.size());
-        for (V v : values) {
-            result.add(convertor.apply(v));
-        }
-        return result;
-    }
-
-    protected abstract List<V> readFromSegments(S segments, Filters<V> filters) throws IOException;
+    protected abstract <R> List<R> readFromSegments(
+            S segments, Filters<V> filters, Function<V, R> convertor) throws IOException;
 
     protected abstract S createSegments(K k, @Nullable Long fileSize);
 

paimon-core/src/main/java/org/apache/paimon/utils/SimpleObjectsCache.java

@@ -37,6 +37,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Function;
 
 /** Cache records to {@link SegmentsCache} by compacted serializer. */
 @ThreadSafe
@@ -58,8 +59,10 @@ public List<V> read(
     }
 
     @Override
-    protected List<V> readFromSegments(Segments segments, Filters<V> filters) throws IOException {
-        return readFromSegments(formatSerializer.get(), projectedSerializer, segments, filters);
+    protected <R> List<R> readFromSegments(
+            Segments segments, Filters<V> filters, Function<V, R> convertor) throws IOException {
+        return readFromSegments(
+                formatSerializer.get(), projectedSerializer, segments, filters, convertor);
     }
 
     @Override
@@ -81,13 +84,14 @@ protected Segments createSegments(K key, @Nullable Long fileSize) {
         }
     }
 
-    public static <V> List<V> readFromSegments(
+    public static <V, R> List<R> readFromSegments(
             InternalRowSerializer formatSerializer,
             ObjectSerializer<V> projectedSerializer,
             Segments segments,
-            Filters<V> filters)
+            Filters<V> filters,
+            Function<V, R> convertor)
             throws IOException {
-        List<V> entries = new ArrayList<>();
+        List<R> entries = new ArrayList<>();
         RandomAccessInputView view = createInputView(segments);
         BinaryRow binaryRow = new BinaryRow(formatSerializer.getArity());
         Filter<InternalRow> readFilter = filters.readFilter();
@@ -98,7 +102,7 @@ public static <V> List<V> readFromSegments(
                 if (readFilter.test(binaryRow)) {
                     V v = projectedSerializer.fromRow(binaryRow);
                     if (readVFilter.test(v)) {
-                        entries.add(v);
+                        entries.add(convertor.apply(v));
                     }
                 }
             } catch (EOFException e) {