JingsongLi commented on code in PR #9473:
URL: https://github.com/apache/paimon/pull/9473#discussion_r3889460795
##########
paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java:
##########
@@ -159,7 +160,16 @@ public FileRecordReader<InternalRow> createReader(Context
context) throws IOExce
SeekableInputStream in = fileIO.newInputStream(filePath);
Review Comment:
[P2] Avoid opening cached descriptor-only BLOBs
For a scalar BLOB read with blob-as-descriptor=true, a cached BlobFileMeta
needs no payload stream because
RawBlobElementSerializer.requiresReadInputStream returns false. This
nevertheless opens the file before consulting the cache, and
BlobElementSerializer immediately closes that unused stream. A spanning BLOB
reattached to 100 anchors therefore still issues 100 FileSystem.open/close
calls. Please check the cache and serializer requirement before opening, while
retaining reader-local streams for inline and nested BLOB reads.
##########
paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionSplitGenerator.java:
##########
@@ -79,6 +102,67 @@ public List<SplitGroup> splitForBatch(List<DataFileMeta>
input) {
.collect(Collectors.toList());
}
+ private List<List<List<DataFileMeta>>> packWithUniqueSidecars(
+ List<List<DataFileMeta>> ranges, Set<DataFileMeta> sharedSidecars)
{
+ List<List<List<DataFileMeta>>> packed = new ArrayList<>();
+ List<List<DataFileMeta>> current = new ArrayList<>();
+ Set<DataFileMeta> seenSidecars = Collections.newSetFromMap(new
IdentityHashMap<>());
+ long currentWeight = 0;
+
+ for (List<DataFileMeta> range : ranges) {
+ long weight = incrementalRangeWeight(range, seenSidecars,
sharedSidecars);
+ if (!current.isEmpty() && currentWeight + weight >
targetSplitSize) {
+ packed.add(current);
+ current = new ArrayList<>();
+ seenSidecars = Collections.newSetFromMap(new
IdentityHashMap<>());
+ currentWeight = 0;
+ weight = incrementalRangeWeight(range, seenSidecars,
sharedSidecars);
+ }
+ current.add(range);
+ currentWeight += weight;
+ range.stream()
+ .filter(DataEvolutionSplitGenerator::isSidecar)
+ .forEach(seenSidecars::add);
+ }
+
+ if (!current.isEmpty()) {
+ packed.add(current);
+ }
+ return packed;
+ }
+
+ private long incrementalRangeWeight(
+ List<DataFileMeta> files,
+ Set<DataFileMeta> seenSidecars,
+ Set<DataFileMeta> sharedSidecars) {
+ Set<DataFileMeta> seenInRange = Collections.newSetFromMap(new
IdentityHashMap<>());
+ long size =
+ files.stream()
+ .filter(
+ file ->
+ !isSidecar(file)
+ ||
(!sharedSidecars.contains(file)
+ &&
!seenSidecars.contains(file)
+ &&
seenInRange.add(file)))
Review Comment:
[P2] Charge each newly encountered shared sidecar
Filtering every shared sidecar out globally handles one oversized file
spanning all anchors, but it also hides distinct shared files with disjoint
coverage. Two 100 MiB sidecars covering anchors [0,1] and [2,3] are packed into
one roughly 200 MiB task with a 150 MiB target, even though they can be
separated without duplicating either sidecar. Please suppress only repeated
references to the same sidecar already present in a candidate pack (with a
special case for an unavoidable oversized fixed cost), while charging a
different shared sidecar when it first enters the pack.
##########
paimon-python/pypaimon/read/reader/format_blob_reader.py:
##########
@@ -74,8 +76,28 @@ def __init__(self, file_io: FileIO, file_path: str,
read_fields: List[str],
if file_size is not None and file_size > 0
else file_io.get_file_size(file_path)
)
- self._input_stream = file_io.new_input_stream(file_path)
- self._read_index()
+ cached_index = (
+ blob_index_cache.get(file_path)
+ if blob_index_cache is not None
+ else None
+ )
+ if cached_index is None:
+ self._input_stream = file_io.new_input_stream(file_path)
+ self._read_index()
+ if blob_index_cache is not None:
+ if self._is_video:
+ blob_index_cache[file_path] = copy(self._video_meta)
+ else:
+ cached_index = (
+ tuple(self.blob_lengths), tuple(self.blob_offsets)
+ )
+ blob_index_cache[file_path] = cached_index
+ self.blob_lengths, self.blob_offsets = cached_index
+ elif self._is_video:
+ self._video_meta = copy(cached_index)
+ else:
+ self._input_stream = file_io.new_input_stream(file_path)
Review Comment:
[P2] Avoid unused BLOB opens on cache hits
A cached ordinary BLOB index still opens a reader-local stream here before
the field shape and read mode are known. Scalar descriptor reads (the
multimodal default) and concurrent scalar reads immediately close that stream
below without reading it, so a spanning BLOB attached to N anchors still
performs N remote open_input_file calls even though its index is parsed once.
Please defer opening until the mode is known and skip it for those
descriptor/concurrent cases, while retaining streams for inline and nested BLOB
reads.
##########
paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py:
##########
@@ -112,6 +128,117 @@ def weight_func(file_list: List[DataFileMeta]) -> int:
return splits
+ def _pack_with_unique_sidecars(
+ self, groups: List[List[DataFileMeta]]
+ ) -> List[List[List[DataFileMeta]]]:
+ sidecar_occurrences = defaultdict(int)
+ for group in groups:
+ for identity in {
+ id(file) for file in group if self._is_sidecar(file)
+ }:
+ sidecar_occurrences[identity] += 1
+ shared_sidecars = {
+ identity for identity, count in sidecar_occurrences.items()
+ if count > 1
+ }
+
+ packed = []
+ current = []
+ current_weight = 0
+ seen_sidecars = set()
+
+ for group in groups:
+ weight = self._incremental_group_weight(
+ group, seen_sidecars, shared_sidecars
+ )
+ if current and current_weight + weight > self.target_split_size:
+ packed.append(current)
+ current = []
+ current_weight = 0
+ seen_sidecars = set()
+ weight = self._incremental_group_weight(
+ group, seen_sidecars, shared_sidecars
+ )
+
+ current.append(group)
+ current_weight += weight
+ seen_sidecars.update(
+ id(file) for file in group if self._is_sidecar(file)
+ )
+
+ if current:
+ packed.append(current)
+ return packed
+
+ def _incremental_group_weight(
+ self, group: List[DataFileMeta], seen_sidecars: set,
+ shared_sidecars: set,
+ ) -> int:
+ seen_in_group = set()
+ size = 0
+ for file in group:
+ if self._is_sidecar(file):
+ identity = id(file)
+ if identity in shared_sidecars:
Review Comment:
[P2] Keep distinct shared sidecars in the packing weight
This assigns zero weight to every sidecar that spans multiple anchors,
including independent files with disjoint coverage. With ten 64 MiB videos each
spanning its own pair of tiny anchors and a 130 MiB target, current head
produces one roughly 640 MiB split instead of five roughly 128 MiB splits.
Please special-case only the unavoidable oversized shared cost and continue
charging each distinct shared sidecar once per candidate pack.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
