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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
