You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org> on 2026/08/30 04:49:58 UTC
[PR] [core][python] Support multiple packed video fields [paimon]
XiaoHongbo-Hope opened a new pull request, #9473:
URL: https://github.com/apache/paimon/pull/9473
## What
- Allow `video-frame-field` to configure multiple scalar BLOB fields.
- Roll each `.video` field independently, so different camera boundaries do not split or duplicate encoded videos.
- Align independently rolled sidecars with normal-file row ranges in Java and Python read/commit paths.
## Tests
- Java: two video fields with different boundaries, schema validation, split generation, and row-range reads.
- Python: independent video rolling, commit row IDs, retry safety, and multimodal column selection.
- Local checks: Java compile/Spotless, 5 targeted Java tests, 86 targeted Python tests, Flake8, `py_compile`, and `git diff --check`.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][python] Support multiple packed video fields [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][python] Support multiple packed video fields [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9473:
URL: https://github.com/apache/paimon/pull/9473#discussion_r3889309588
##########
paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionSplitGenerator.java:
##########
@@ -79,6 +86,62 @@ public List<SplitGroup> splitForBatch(List<DataFileMeta> input) {
.collect(Collectors.toList());
}
+ private List<List<List<DataFileMeta>>> packWithUniqueSidecars(List<List<DataFileMeta>> ranges) {
+ 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);
+ if (!current.isEmpty() && currentWeight + weight > targetSplitSize) {
Review Comment:
[P2] Keep an oversized shared sidecar from forcing one split per anchor. When the first spanning sidecar already exceeds targetSplitSize, currentWeight starts above the target. The next small anchor therefore flushes the bin, resets seenSidecars, and counts the same oversized sidecar again; this repeats for every anchor (100 ten-byte normal ranges plus one 100 KB spanning video with a 500-byte target produce 100 splits). Oversized videos are expected because VideoRollingFileWriter deliberately treats its target as soft to keep a physical video intact. Please separate the unavoidable shared fixed cost from marginal per-range weight so covered anchors can still coalesce.
##########
paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py:
##########
@@ -112,6 +128,99 @@ def weight_func(file_list: List[DataFileMeta]) -> int:
return splits
+ def _pack_with_unique_sidecars(
+ self, groups: List[List[DataFileMeta]]
+ ) -> List[List[List[DataFileMeta]]]:
+ packed = []
+ current = []
+ current_weight = 0
+ seen_sidecars = set()
+
+ for group in groups:
+ weight = self._incremental_group_weight(group, seen_sidecars)
+ if current and current_weight + weight > self.target_split_size:
Review Comment:
[P2] Keep an oversized shared sidecar from forcing one split per anchor. If the first spanning sidecar already exceeds target_split_size, current_weight remains above the target. The next small anchor flushes the pack, resets seen_sidecars, and counts the same oversized sidecar again, so 100 tiny anchors plus one shared 100 KB video with a 500-byte target produce 100 splits. Oversized videos are valid because a physical video group cannot be cut at the configured target. Please separate this unavoidable shared fixed cost from marginal per-group weight so its covered anchors can still be coalesced.
##########
paimon-python/pypaimon/read/reader/format_blob_reader.py:
##########
@@ -74,8 +76,18 @@ 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_video_meta = (
+ video_meta_cache.get(file_path)
+ if self._is_video and video_meta_cache is not None
Review Comment:
[P2] Cache ordinary BLOB indexes as well. This cache is gated on _is_video, but .blob writers also roll independently from normal files and blob compaction can emit one .blob covering many normal anchors. DataEvolutionSplitRead then constructs one FormatBlobReader per anchor, so each takes the miss path and reopens and decompresses the full delta-varint index. I reproduced three Parquet anchors plus one 3,000-row .blob opening and parsing the same .blob three times in one packed split. Please cache immutable unselected blob_lengths and blob_offsets too, while keeping payload streams and row selections reader-local.
##########
paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java:
##########
@@ -1231,7 +1242,7 @@ public static List<List<DataFileMeta>> mergeRangesAndSort(List<DataFileMeta> fil
// group by row id range
ToLongFunction<DataFileMeta> maxSeqF = DataFileMeta::maxSequenceNumber;
RangeHelper<DataFileMeta> rangeHelper = new RangeHelper<>(DataFileMeta::nonNullRowIdRange);
- List<List<DataFileMeta>> result = rangeHelper.mergeOverlappingRanges(files);
+ List<List<DataFileMeta>> result = groupByNormalFileRange(files, Function.identity());
Review Comment:
[P2] Reuse sidecar metadata across anchor readers. groupByNormalFileRange reattaches a spanning BLOB or video to every normal anchor, and createReader builds a fresh union-reader supplier for every group. Each supplier reaches BlobFileFormat or VideoFileFormat, which reopens the physical file and rebuilds its complete BlobFileMeta or VideoFileMeta; one packed split with 100 anchors therefore performs 100 remote opens and full index parses inside one task. Please keep an immutable per-split metadata cache keyed by physical sidecar and derive each anchor selection from it, or otherwise initialize each sidecar once.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][python] Support multiple packed video fields [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9473:
URL: https://github.com/apache/paimon/pull/9473#discussion_r3889619057
##########
paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java:
##########
@@ -201,4 +202,105 @@ public static Range checkContiguousRowRange(List<DataFileMeta> files) {
merged);
return merged.get(0);
}
+
+ /**
+ * Groups data-evolution files by normal-file row ranges.
+ *
+ * <p>Blob and vector files may span several adjacent normal files. Such sidecar files are
+ * included in every anchor group they intersect, so each group can select just its normal-file
+ * range without duplicating the physical sidecar.
+ */
+ public static <T> List<List<T>> groupByNormalFileRange(
Review Comment:
[Suggestion] Model an anchored range explicitly instead of returning nested lists. The important contract here is that a sidecar may be attached to several anchor groups as the same object identity, while an unanchored group has different semantics. That contract is currently erased by `List<List<T>>`, so scan, split generation, `DataSplit`, and split read each have to rediscover anchors, shared-sidecar occurrences, and identity deduplication. A small `NormalRangeGroup<T>` (anchor range, normal entries, intersecting sidecars, and unanchored state) would put coverage ownership in one place and make the downstream code substantially easier to audit; a flattening adapter could keep callers incremental.
##########
paimon-common/src/main/java/org/apache/paimon/format/FormatReaderFactory.java:
##########
@@ -60,5 +61,11 @@ interface Context {
default ReadBatchSizer readBatchSizer() {
return null;
}
+
+ /** Metadata shared by readers of the same physical file within one split. */
+ @Nullable
+ default Map<Path, Object> metadataCache() {
Review Comment:
[Suggestion] Hide this untyped map behind a metadata-cache abstraction. Adding `Map<Path, Object>` to the general format SPI makes cache lifetime and value type an implicit cross-module contract, and BLOB/video readers now duplicate unchecked get/cast/load/put logic. A typed `getOrLoad(Path, Class<T>, loader)` (or format-scoped cache key) owned by `FormatReaderContext` would centralize miss/error behavior, prevent accidental type collisions, and let `DataEvolutionSplitRead` pass a small per-split read session instead of plumbing a mutable raw map through its reader-building methods.
##########
paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionSplitGenerator.java:
##########
@@ -79,6 +102,111 @@ public List<SplitGroup> splitForBatch(List<DataFileMeta> input) {
.collect(Collectors.toList());
}
+ private List<List<List<DataFileMeta>>> packWithUniqueSidecars(
Review Comment:
[Suggestion] Encapsulate the packing state and make the weight calculation side-effect free. `current`, `currentWeight`, `seenSidecars`, and `fixedComponent` must move together, but the full reset is hand-written in two branches and `initialRangeWeight` mutates `fixedComponent` as an output parameter. A private `PackingState` with one finish/reset transition plus an `InitialWeight { weight, fixedSidecars }` result would make the coverage invariant visible and reduce the chance that the Java and Python mirror implementations drift during the next edge-case change.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][python] Support multiple packed video fields [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9473:
URL: https://github.com/apache/paimon/pull/9473#discussion_r3889040441
##########
paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java:
##########
@@ -114,9 +115,10 @@ public MultipleBlobFileWriter(
null);
RollingFileWriterImpl<InternalRow, DataFileMeta> rollingWriter =
video
- ? new VideoRollingFileWriter<>(writerFactory, targetFileSize)
+ ? new VideoRollingFileWriter<>(
+ writerFactory, targetFileSize, targetFileRowNum)
Review Comment:
[P2] Passing targetFileRowNum to VideoRollingFileWriter exposes a NULL-boundary bug in its deferred policy. Consecutive SQL NULL video values produce currentVideo == nextVideo == null, so after a target sets pendingRoll, Objects.equals(null, null) prevents the file from ever closing. The previous outer writer closed all writers immediately when its target was reached on a NULL group. Please treat NULL as having no protected physical-video group and close once rolling is pending; an all-NULL row-cap regression test would cover this.
##########
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) {
+ return ranges.stream()
+ .map(SplitGroup::nonRawConvertibleGroup)
Review Comment:
[P1] A spanning sidecar is now inserted into every intersecting anchor split, but DataSplit.dataEvolutionMergedRowCount() still uses the maximum physical file row count in each overlap group. In the new six-row/two-camera layout, the three two-row anchor splits each contain a four-row video file, so they report 4 + 4 + 4 = 12 logical rows. Core LIMIT pushdown can consequently omit required splits, and Flink COUNT(*) pushdown can return 12 instead of 6. Please derive the count from the normal anchor range, or carry the effective logical range explicitly in the split.
##########
paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java:
##########
@@ -372,14 +364,9 @@ public void write(InternalRow row) throws IOException {
}
recordCount++;
currentFileRecordCount++;
- currentVideoGroup = nextVideoGroup;
if (rollingFile()) {
- if (nextVideoGroup == null) {
- closeCurrentWriter();
- } else {
- pendingGroupAwareRoll = true;
- }
+ closeNormalAndVectorWriters();
Review Comment:
[P2] Once blob/video files roll independently here, one sidecar can bridge many normal-file ranges. DataEvolutionFileStoreScan.postFilterManifestEntries still groups all files with RangeHelper, so that sidecar connects the normal ranges into one large stats group. The normal providers then cover only a subset of the group and are marked unknown, disabling otherwise selective normal-column stats pruning and forcing a whole-video scan. Please compute evolution stats per normal anchor range, excluding sidecar extents, and deduplicate shared sidecars after filtering.
##########
paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py:
##########
@@ -80,25 +77,43 @@ def weight_func(file_list: List[DataFileMeta]) -> int:
group for group in split_by_row_id
if self.group_stats_filter.may_match(group)
]
+
+ file_ids = [id(file) for group in split_by_row_id for file in group]
+ has_spanning_sidecar = len(file_ids) != len(set(file_ids))
+ if self.group_stats_filter is not None:
split_by_row_id = [
[file.copy_without_stats() 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
- )
+ if has_spanning_sidecar:
Review Comment:
[P2] has_spanning_sidecar is partition-wide, so one spanning sidecar bypasses ordered bin packing for every normal range in the partition. A 100-frame video with target-file-row-num=1 produces 100 splits even when target_split_size is large enough to pack them, creating one task per frame range and repeatedly opening the same sidecar. Please continue packing normal anchor groups and deduplicate shared sidecar metadata within each packed DataSplit, or otherwise limit the unpacking to groups that truly cannot be combined.
##########
paimon-python/pypaimon/write/writer/dedicated_format_writer.py:
##########
@@ -568,14 +517,37 @@ def pending_row_count(self) -> int:
return 0
def _close_current_writers(self):
- """Close normal, blob, and vector writers; add metadata in order: normal, blob, vector."""
- # A flush spans the normal file and every blob/vector sidecar, and the
- # sidecar writers drain their own buffers as they go, so their half cannot
- # be replayed from scratch. Two rules make a retry resume rather than
- # restart: the normal rows stay buffered until their file lands, and once
- # it has landed the file is remembered instead of the rows. Nothing
- # reaches ``committed_files`` until every phase has succeeded, so a retry
- # never finds a half-published flush.
+ """Close normal/vector writers, then finalize independently rolling blobs."""
+ self._close_normal_and_vector_writers()
+ if self._blob_writers_closed:
Review Comment:
[P1] This flag is lifetime-scoped, while StreamTableWrite is reusable across prepare_commit calls. After the first prepare sets _blob_writers_closed, later writes do not reset it, so the next prepare returns here after publishing the new normal/vector metadata and never harvests the new blob/video metadata. A write -> prepare -> write -> prepare reproduction committed the second normal file but returned NULL for its blob and left the sidecar unreferenced. Please make blob finalization checkpoint/epoch-scoped; note that simply resetting the flag is insufficient because total_record_count is currently cumulative while subwriter metadata is cleared after each harvest.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][python] Support multiple packed video fields [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][python] Support multiple packed video fields [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #9473:
URL: https://github.com/apache/paimon/pull/9473
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][python] Support multiple packed video fields [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9473:
URL: https://github.com/apache/paimon/pull/9473#discussion_r3889529997
##########
paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionSplitGenerator.java:
##########
@@ -79,6 +102,73 @@ 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, Collections.emptySet());
+ if (current.isEmpty() && weight > targetSplitSize) {
+ weight = incrementalRangeWeight(range, seenSidecars, sharedSidecars);
Review Comment:
[P2] Close an oversized sidecar component before charging a new one
After an oversized shared sidecar is ignored here, currentWeight retains only the tiny anchor cost. A later independent sidecar can therefore appear to fit and be appended to the same split: a 200 MiB sidecar spanning ranges 0-1 followed by a 100 MiB sidecar spanning ranges 2-3 produces one roughly 300 MiB task with a 128 MiB target. The boundary between these coverage components can be split without rereading either file. Please keep coalescing ranges covered by the ignored sidecar, but flush the saturated component before admitting a newly encountered independent sidecar.
##########
paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py:
##########
@@ -112,6 +128,125 @@ 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, set()
+ )
+ if not current and weight > self.target_split_size:
+ weight = self._incremental_group_weight(
Review Comment:
[P2] End the oversized-sidecar exemption at its coverage boundary
This removes the first oversized shared sidecar from current_weight for the lifetime of the pack. Once that sidecar no longer covers subsequent anchors, an unrelated sidecar can still appear to fit: with a 128 MiB target, a 200 MiB sidecar over anchors 0-1 plus an independent 100 MiB sidecar over 2-3 becomes one roughly 300 MiB split, although they can be separate tasks without duplicating either file. Please track the ignored sidecar component and flush before admitting an unrelated sidecar after its coverage ends.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][python] Support multiple packed video fields [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9473:
URL: https://github.com/apache/paimon/pull/9473#discussion_r3889039633
##########
paimon-core/src/main/java/org/apache/paimon/operation/commit/RowTrackingCommitUtils.java:
##########
@@ -95,7 +111,7 @@ private static long assignRowTrackingMeta(
long rowCount = entry.file().rowCount();
if (isBlobFile(entry.file().fileName())) {
String blobFieldName = entry.file().writeCols().get(0);
- long blobStart = blobStarts.getOrDefault(blobFieldName, blobStartDefault);
+ long blobStart = blobStarts.getOrDefault(blobFieldName, firstRowIdStart);
Review Comment:
[P1] This two-pass assignment loses the normal-range base of each CommitMessage. If a normal-only partial writer is followed in the same commit by a normal+blob writer, the first blob cursor still defaults to the commit-wide firstRowIdStart, so the blob from the second writer is assigned to the first logical row. I reproduced this as payloads [belongs-to-row-2, null] instead of [null, belongs-to-row-2]. Please preserve the writer/message boundary when allocating sidecar row IDs, or carry an explicit relative row offset. The mirrored Python implementation in file_store_commit.py has the same issue.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org