JingsongLi commented on code in PR #9473:
URL: https://github.com/apache/paimon/pull/9473#discussion_r3889253292
##########
paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py:
##########
@@ -80,25 +77,44 @@ def weight_func(file_list: List[DataFileMeta]) -> int:
group for group in split_by_row_id
if self.group_stats_filter.may_match(group)
]
+
+ has_spanning_sidecar = self._has_spanning_sidecar(split_by_row_id)
+ if self.group_stats_filter is not None:
+ copies = {}
+
+ def without_stats(file):
+ copy = copies.get(id(file))
+ if copy is None:
+ copy = file.copy_without_stats()
+ copies[id(file)] = copy
+ return copy
+
split_by_row_id = [
- [file.copy_without_stats() for file in group]
+ [without_stats(file) for file in group]
for group in split_by_row_id
]
- # Pack the split groups for optimal split sizes
packed_files = self._pack_for_ordered(
- split_by_row_id, weight_func, self.target_split_size
+ split_by_row_id,
+ self._normal_group_weight,
+ self.target_split_size,
)
# Flatten the packed files and build splits
flatten_packed_files: List[List[DataFileMeta]] = [
- [file for sub_pack in pack for file in sub_pack]
+ self._unique_files(pack)
for pack in packed_files
]
- splits += self._build_split_from_pack_for_data_evolution(
+ new_splits = self._build_split_from_pack_for_data_evolution(
flatten_packed_files, packed_files, entries_list
)
+ if has_spanning_sidecar and slice_row_ranges is None and
self.row_ranges is None:
+ new_splits = [
+ IndexedSplit(split, self._normal_ranges(pack))
Review Comment:
[P1] Preserve the DV-aware merged row count in this wrapper.
IndexedSplit.merged_row_count() falls back to the raw row-range cardinality
when exact_merged_row_count is omitted. With two one-row anchors, one spanning
video, and a cardinality-1 deletion vector, this new wrapper reports 2 while
the underlying DataSplit reports 1; if DV cardinality is unavailable, it
similarly converts an unknown count into a known physical count. This makes
split and Ray row-count metadata incorrect. Please carry the underlying exact
count here and preserve the unknown state, with regressions for both known and
unknown DV cardinalities.
##########
paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py:
##########
@@ -112,6 +128,62 @@ def weight_func(file_list: List[DataFileMeta]) -> int:
return splits
+ def _normal_group_weight(self, group: List[DataFileMeta]) -> int:
+ normal_files = [
+ file for file in group
+ if (not DataFileMeta.is_blob_file(file.file_name)
+ and not DataFileMeta.is_vector_file(file.file_name))
+ ]
+ files = normal_files or group
+ return max(sum(file.file_size for file in files), self.open_file_cost)
Review Comment:
[P2] Keep distinct sidecar sizes in the packing weight. This removes every
BLOB or vector file from the weight, not only repeated references to one
spanning sidecar. For example, ten one-range 64 MiB video files plus tiny
normal anchors are now packed into one approximately 640 MiB split even with a
128 MiB target. That makes source.split.target-size ineffective and collapses
scan parallelism for multi-video data. Please charge each unique sidecar once
per packed split while avoiding only the duplicate charge for spanning
references.
##########
paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionSplitGenerator.java:
##########
@@ -49,8 +53,15 @@ public boolean alwaysRawConvertible() {
@Override
public List<SplitGroup> splitForBatch(List<DataFileMeta> input) {
- RangeHelper<DataFileMeta> rangeHelper = new
RangeHelper<>(DataFileMeta::nonNullRowIdRange);
- List<List<DataFileMeta>> ranges =
rangeHelper.mergeOverlappingRanges(input);
+ List<List<DataFileMeta>> ranges = groupByNormalFileRange(input,
Function.identity());
+ Set<DataFileMeta> seen = Collections.newSetFromMap(new
IdentityHashMap<>());
+ boolean hasSpanningSidecar =
+ ranges.stream().flatMap(Collection::stream).anyMatch(file ->
!seen.add(file));
+ if (hasSpanningSidecar) {
Review Comment:
[P2] Preserve bin packing when a sidecar spans anchor ranges. This
partition-wide early return turns every normal anchor into a separate split:
with target-file-row-num=1, a 100-frame sidecar produces 100 tasks and reopens
the same video file in each one even when targetSplitSize could pack all
ranges. Please pack the normal-anchor groups, deduplicate shared sidecar
metadata within each packed split, and carry the effective anchor ranges to the
reader.
##########
paimon-python/pypaimon/read/split_read.py:
##########
@@ -1274,49 +1275,26 @@ def _create_prescan_reader(self, field_names):
return prescan_read._create_raw_reader()
def _split_by_row_id(self, files: List[DataFileMeta]) ->
List[List[DataFileMeta]]:
- """Split files by firstRowId for data evolution."""
-
- # Sort files by firstRowId and then by maxSequenceNumber
- def sort_key(file: DataFileMeta) -> tuple:
- first_row_id = file.first_row_id if file.first_row_id is not None
else float('-inf')
- is_special = 1 if (DataFileMeta.is_blob_file(file.file_name)
- or DataFileMeta.is_vector_file(file.file_name))
else 0
- max_seq = file.max_sequence_number
- return (first_row_id, is_special, -max_seq)
-
- sorted_files = sorted(files, key=sort_key)
-
- # Split files by firstRowId
- split_by_row_id = []
- last_row_id = -1
- check_row_id_start = 0
- current_split = []
-
- for file in sorted_files:
- first_row_id = file.first_row_id
- if first_row_id is None:
- split_by_row_id.append([file])
- continue
-
- if (not DataFileMeta.is_blob_file(file.file_name)
- and not DataFileMeta.is_vector_file(file.file_name)
- and first_row_id != last_row_id):
- if current_split:
- split_by_row_id.append(current_split)
- if first_row_id < check_row_id_start:
- raise ValueError(
- f"There are overlapping files in the split: {files}, "
- f"the wrong file is: {file}"
- )
- current_split = []
- last_row_id = first_row_id
- check_row_id_start = first_row_id + file.row_count
- current_split.append(file)
-
- if current_split:
- split_by_row_id.append(current_split)
-
- return split_by_row_id
+ """Group sidecars by every normal-file range they intersect."""
+ files_without_row_id = [file for file in files if file.first_row_id is
None]
+ ranged_files = [file for file in files if file.first_row_id is not
None]
+ groups = [[file] for file in files_without_row_id]
+ groups.extend(group_by_normal_file_range(
Review Comment:
[P2] Packing and deduplicating metadata does not prevent the sidecar from
being reopened here. _create_raw_reader later creates one union reader per
reconstructed normal-range group, and each group constructs a new
FormatBlobReader for the same file. I reproduced one packed split with 10
one-row normal files and one video file opening the video reader 10 times.
Please share one sidecar reader or index across the packed logical ranges, or
preserve enough grouping metadata to avoid rebuilding it per anchor.
--
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]
