From d5d427cd2c79c212ba11d88e088991d91d90b56e Mon Sep 17 00:00:00 2001 From: zhoulii Date: Tue, 25 Aug 2026 16:55:16 +0800 Subject: [PATCH] [core][python] Fix partial ranges when updating newly added Blob columns --- .../operation/BlobFallbackRecordReader.java | 51 +++++++++++- .../operation/DataEvolutionSplitRead.java | 65 +++++++++++---- .../BlobFallbackRecordReaderTest.java | 44 ++++++++++ .../operation/DataEvolutionReadTest.java | 23 ++++- .../read/reader/concat_batch_reader.py | 30 ++++--- .../pypaimon/read/reader/field_bunch.py | 58 +++++++++---- paimon-python/pypaimon/read/split_read.py | 31 +++++-- .../pypaimon/tests/blob_table_test.py | 83 +++++++++++++++++++ paimon-python/pypaimon/tests/blob_test.py | 47 +++++++++++ .../pypaimon/tests/field_bunch_test.py | 28 ++++++- .../pypaimon/write/table_update_by_row_id.py | 62 +++++++++----- 11 files changed, 441 insertions(+), 81 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java b/paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java index db501c1615a1..927933fce001 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java @@ -77,6 +77,25 @@ public class BlobFallbackRecordReader implements RecordReader { RowType readRowType, int blobIndex) throws IOException { + this( + files, + readerFactory, + readerWrapper, + enclosingRange(files), + rowRanges, + readRowType, + blobIndex); + } + + BlobFallbackRecordReader( + List files, + BlobFileReaderFactory readerFactory, + ReaderWrapper readerWrapper, + Range logicalRange, + List rowRanges, + RowType readRowType, + int blobIndex) + throws IOException { this.fieldCount = readRowType.getFieldCount(); this.rowIdIndex = readRowType.getFieldIndex(SpecialFields.ROW_ID.name()); this.seqNumIndex = readRowType.getFieldIndex(SpecialFields.SEQUENCE_NUMBER.name()); @@ -85,15 +104,27 @@ public class BlobFallbackRecordReader implements RecordReader { InternalRow.createFieldGetter(readRowType.getTypeAt(blobIndex), blobIndex); checkArgument(!files.isEmpty(), "Blob bunch should not be empty."); - long firstRowId = Long.MAX_VALUE; - long lastRowId = Long.MIN_VALUE; + long firstRowId = logicalRange.from; + long lastRowId = logicalRange.to; // sort group readers in descending order Map> sequenceGroups = new TreeMap<>(reverseOrder()); for (DataFileMeta file : files) { Range fileRange = file.nonNullRowIdRange(); - firstRowId = Math.min(firstRowId, fileRange.from); - lastRowId = Math.max(lastRowId, fileRange.to); + if (rowRanges == null) { + checkArgument( + fileRange.from >= firstRowId && fileRange.to <= lastRowId, + "Blob file range %s should be within logical range %s.", + fileRange, + logicalRange); + } else { + // A pushed range may select one normal-file range from a spanning Blob file. + checkArgument( + fileRange.hasIntersection(logicalRange), + "Blob file range %s should intersect logical range %s.", + fileRange, + logicalRange); + } sequenceGroups .computeIfAbsent(file.maxSequenceNumber(), ignored -> new ArrayList<>()) @@ -132,6 +163,18 @@ public class BlobFallbackRecordReader implements RecordReader { } } + private static Range enclosingRange(List files) { + checkArgument(!files.isEmpty(), "Blob bunch should not be empty."); + long firstRowId = Long.MAX_VALUE; + long lastRowId = Long.MIN_VALUE; + for (DataFileMeta file : files) { + Range range = file.nonNullRowIdRange(); + firstRowId = Math.min(firstRowId, range.from); + lastRowId = Math.max(lastRowId, range.to); + } + return new Range(firstRowId, lastRowId); + } + @Nullable @Override public RecordIterator readBatch() throws IOException { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java index e064736419db..2a606ee677eb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java @@ -306,7 +306,7 @@ private DataEvolutionFileReader createUnionReader( rowRanges != null); long rowCount = fieldsFiles.get(0).rowCount(); - long firstRowId = fieldsFiles.get(0).files().get(0).nonNullFirstRowId(); + long firstRowId = bunchFirstRowId(fieldsFiles.get(0)); if (rowRanges == null) { for (FieldBunch bunch : fieldsFiles) { @@ -314,7 +314,7 @@ private DataEvolutionFileReader createUnionReader( bunch.rowCount() == rowCount, "All files in a field merge split should have the same row count."); checkArgument( - bunch.files().get(0).nonNullFirstRowId() == firstRowId, + bunchFirstRowId(bunch) == firstRowId, "All files in a field merge split should have the same first row id and could not be null."); } } @@ -437,7 +437,8 @@ private RecordReader createFieldBunchReader( // for blob bunch, fallback on placeholders // fast path: only contains one max_seq group - if (((BlobFileBunch) bunch).sequentialReadOptimize()) { + BlobFileBunch blobBunch = (BlobFileBunch) bunch; + if (blobBunch.sequentialReadOptimize()) { return sequentialReadFiles( bunch.files(), partition, @@ -461,6 +462,7 @@ private RecordReader createFieldBunchReader( deletionVector), (reader, range) -> applyDeletionVector(reader, range, rowRanges, deletionVector), + blobBunch.logicalRange(), rowRanges, readRowType, blobIndex); @@ -1008,14 +1010,15 @@ public static List splitFieldBunches( Map blobBunchMap = new HashMap<>(); Map vectorStoreBunchMap = new TreeMap<>(); long rowCount = -1; + Range rowRange = null; for (DataFileMeta file : needMergeFiles) { if (isBlobFile(file.fileName())) { RowType rowType = fileToRowType.apply(file); int fieldId = rowType.getField(file.writeCols().get(0)).id(); - final long expectedRowCount = rowCount; + final Range expectedRowRange = rowRange; blobBunchMap .computeIfAbsent( - fieldId, key -> new BlobFileBunch(expectedRowCount, rowIdPushDown)) + fieldId, key -> new BlobFileBunch(expectedRowRange, rowIdPushDown)) .add(file); } else if (isVectorStoreFile(file.fileName())) { RowType rowType = fileToRowType.apply(file); @@ -1033,6 +1036,7 @@ public static List splitFieldBunches( // Normal file, just add it to the current merge split fieldsFiles.add(new DataBunch(file)); rowCount = file.rowCount(); + rowRange = file.nonNullRowIdRange(); } } fieldsFiles.addAll(blobBunchMap.values()); @@ -1048,6 +1052,13 @@ public interface FieldBunch { List files(); } + private static long bunchFirstRowId(FieldBunch bunch) { + if (bunch instanceof BlobFileBunch) { + return ((BlobFileBunch) bunch).logicalRange().from; + } + return bunch.files().get(0).nonNullFirstRowId(); + } + private static class DataBunch implements FieldBunch { private final DataFileMeta dataFile; @@ -1076,12 +1087,14 @@ static class BlobFileBunch implements FieldBunch { final List files; final List ranges; - final long expectedRowCount; + // The normal file owns the logical rows; a Blob column added later may physically cover + // only a subset of that anchor range. + @Nullable final Range expectedRowRange; final boolean rowIdPushdown; - BlobFileBunch(long expectedRowCount, boolean rowIdPushdown) { + BlobFileBunch(@Nullable Range expectedRowRange, boolean rowIdPushdown) { this.files = new ArrayList<>(); - this.expectedRowCount = expectedRowCount; + this.expectedRowRange = expectedRowRange; this.ranges = new ArrayList<>(); this.rowIdPushdown = rowIdPushdown; } @@ -1103,24 +1116,40 @@ void add(DataFileMeta file) { @Override public long rowCount() { List merged = Range.sortAndMergeOverlap(ranges, true); + if (expectedRowRange != null) { + for (Range range : merged) { + Preconditions.checkState( + range.from >= expectedRowRange.from && range.to <= expectedRowRange.to, + "Blob file range %s should be within normal file range %s.", + range, + expectedRowRange); + } + return expectedRowRange.count(); + } + if (!rowIdPushdown) { Preconditions.checkState( merged.size() == 1, "Blob file bunch should always contain a contiguous row range."); - - long rowCount = merged.get(0).count(); - if (expectedRowCount >= 0) { - Preconditions.checkState( - rowCount == expectedRowCount, - "The merged rowCount %s of blob file bunch should be aligned with normal files %s.", - rowCount, - expectedRowCount); - } } return merged.stream().mapToLong(Range::count).sum(); } + Range logicalRange() { + if (expectedRowRange != null) { + return expectedRowRange; + } + List merged = Range.sortAndMergeOverlap(ranges, true); + Preconditions.checkState(!merged.isEmpty(), "Blob file bunch should not be empty."); + return new Range(merged.get(0).from, merged.get(merged.size() - 1).to); + } + + private boolean fullyCoversLogicalRange() { + List merged = Range.sortAndMergeOverlap(ranges, true); + return merged.size() == 1 && merged.get(0).equals(logicalRange()); + } + public boolean sequentialReadOptimize() { Preconditions.checkState(!files.isEmpty(), "Blob file bunch should not be empty."); @@ -1133,7 +1162,7 @@ public boolean sequentialReadOptimize() { } } - return true; + return fullyCoversLogicalRange(); } @Override diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java index 859e8bc5b885..328aaa3061d7 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java @@ -313,6 +313,50 @@ public void testBlobFallbackRecordReaderDerivesRowIdBoundsFromFiles() throws Exc assertThat(rows.sequenceNumbers).containsExactly(1L, 1L, 1L, 2L, 1L, 1L); } + @Test + public void testBlobFallbackRecordReaderFillsLogicalRangeGapsWithNull() throws Exception { + DataFileMeta file = blobFile("partial-file", 2, 2, 1); + + ReadResult rows = + ReadResult.read( + new BlobFallbackRecordReader( + Collections.singletonList(file), + blobFile -> + oneRowPerBatchReader(blobFile, fileRows(blobFile, null)), + (reader, range) -> reader, + new Range(0, 5), + null, + READ_ROW_TYPE, + BLOB_INDEX)); + + assertThat(rows.rowIds).containsExactly(2L, 3L); + assertThat(rows.nullBlobRowIds).containsExactly(0L, 1L, 4L, 5L); + assertThat(rows.nullBlobSequenceNumbers).containsOnly(-1L); + } + + @Test + public void testBlobFallbackRecordReaderClipsSpanningFileToLogicalRange() throws Exception { + DataFileMeta file = blobFile("spanning-file", 5, 10, 1); + List rowRanges = Collections.singletonList(new Range(5, 9)); + + ReadResult rows = + ReadResult.read( + new BlobFallbackRecordReader( + Collections.singletonList(file), + blobFile -> + oneRowPerBatchReader( + blobFile, fileRows(blobFile, rowRanges)), + (reader, range) -> reader, + new Range(5, 9), + rowRanges, + READ_ROW_TYPE, + BLOB_INDEX)); + + assertThat(rows.rowIds).containsExactly(5L, 6L, 7L, 8L, 9L); + assertThat(rows.sequenceNumbers).containsOnly(1L); + assertThat(rows.nullBlobRowIds).isEmpty(); + } + @Test public void testBlobFallbackRecordReaderReturnsNullIfAllRowsArePlaceholders() throws Exception { DataFileMeta newFile = blobFile("new-placeholder-file", 0, 1, 2); diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionReadTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionReadTest.java index 24d299495753..bdb9fcc93515 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionReadTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionReadTest.java @@ -28,6 +28,7 @@ import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Range; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -464,7 +465,7 @@ void testAddVectorFilesWithDifferentSchemaId() { @Test void testAddBlobFilesWithDifferentSchemaId() { - BlobFileBunch blobBunch = new BlobFileBunch(300, false); + BlobFileBunch blobBunch = new BlobFileBunch(new Range(0, 299), false); DataFileMeta blobEntry1 = createBlobFileWithSchema("blob1", 0, 100, 1, 0L); DataFileMeta blobEntry2 = createBlobFileWithSchema("blob2", 100, 200, 1, 1L); @@ -477,6 +478,26 @@ void testAddBlobFilesWithDifferentSchemaId() { assertThat(blobBunch.rowCount()).isEqualTo(300); } + @Test + void testBlobBunchUsesNormalFileRangeForPartialCoverage() { + BlobFileBunch blobBunch = new BlobFileBunch(new Range(100, 109), false); + blobBunch.add(createBlobFile("blob", 103, 5, 1)); + + assertThat(blobBunch.rowCount()).isEqualTo(10); + assertThat(blobBunch.logicalRange()).isEqualTo(new Range(100, 109)); + assertThat(blobBunch.sequentialReadOptimize()).isFalse(); + } + + @Test + void testBlobBunchRejectsRangeOutsideNormalFile() { + BlobFileBunch blobBunch = new BlobFileBunch(new Range(100, 109), false); + blobBunch.add(createBlobFile("blob", 99, 2, 1)); + + assertThatThrownBy(blobBunch::rowCount) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("should be within normal file range"); + } + @Test public void testRowIdPushDown() { VectorFileBunch vectorBunch = new VectorFileBunch(Long.MAX_VALUE, true); diff --git a/paimon-python/pypaimon/read/reader/concat_batch_reader.py b/paimon-python/pypaimon/read/reader/concat_batch_reader.py index 831e41012a95..39b7dac1d607 100644 --- a/paimon-python/pypaimon/read/reader/concat_batch_reader.py +++ b/paimon-python/pypaimon/read/reader/concat_batch_reader.py @@ -294,11 +294,16 @@ class BlobFallbackBatchReader(RecordBatchReader): def __init__(self, file_reader_suppliers: List[Tuple[DataFileMeta, Callable]], field_name: str, output_type, row_ranges: Optional[List[Range]] = None, blob_as_descriptor: bool = False, deletion_vector=None, batch_size: int = 1024, - blob_parallelism: int = 1): + blob_parallelism: int = 1, + logical_ranges: Optional[List[Range]] = None): self._file_reader_suppliers = file_reader_suppliers self._field_name = field_name self._output_type = output_type self._row_ranges = Range.sort_and_merge_overlap(row_ranges) if row_ranges else None + self._logical_ranges = ( + Range.sort_and_merge_overlap(logical_ranges) + if logical_ranges is not None else None + ) self._blob_as_descriptor = blob_as_descriptor self._is_array_blob = pa.types.is_list(output_type) or pa.types.is_large_list(output_type) self._is_map_blob = pa.types.is_map(output_type) @@ -386,14 +391,15 @@ def read_arrow_batch(self) -> Optional[RecordBatch]: if state.selected_range_index >= len(state.selected_ranges): self._close_state_reader(state) - if not groups: - return None - + groups_by_sequence = [ + groups[sequence] + for sequence in sorted(groups.keys(), reverse=True) + ] result = [] for row_id in batch_row_ids: found = False - for max_sequence_number in sorted(groups.keys(), reverse=True): - candidate = groups[max_sequence_number].get(row_id) + for group in groups_by_sequence: + candidate = group.get(row_id) if candidate is None: continue value, is_placeholder = candidate @@ -402,7 +408,7 @@ def read_arrow_batch(self) -> Optional[RecordBatch]: found = True break if not found: - raise ValueError("All blob files at the same row id store a placeholder.") + result.append(None) if resolve_blobs_concurrently: result = self._resolve_selected_blobs(result) @@ -492,10 +498,12 @@ def _resolve_selected_blobs(self, values: List[object]) -> List[object]: return resolved def _compute_target_ranges(self) -> List[Range]: - ranges = Range.sort_and_merge_overlap([ - file.row_id_range() - for file, _ in self._file_reader_suppliers - ]) + ranges = self._logical_ranges + if ranges is None: + ranges = Range.sort_and_merge_overlap([ + file.row_id_range() + for file, _ in self._file_reader_suppliers + ]) if self._row_ranges is not None: ranges = Range.and_(ranges, self._row_ranges) return ranges diff --git a/paimon-python/pypaimon/read/reader/field_bunch.py b/paimon-python/pypaimon/read/reader/field_bunch.py index 5474bf855abb..65ab1af046ed 100644 --- a/paimon-python/pypaimon/read/reader/field_bunch.py +++ b/paimon-python/pypaimon/read/reader/field_bunch.py @@ -22,7 +22,7 @@ supporting both regular data files and blob files. """ from abc import ABC -from typing import List +from typing import List, Optional from pypaimon.manifest.schema.data_file_meta import DataFileMeta from pypaimon.utils.range import Range @@ -144,8 +144,13 @@ def files(self) -> List[DataFileMeta]: class BlobBunch(_SpecialFieldBunch): """Files for partial field (blob files).""" - def __init__(self, expected_row_count: int, row_id_push_down: bool = False): + def __init__(self, expected_row_count: int, row_id_push_down: bool = False, + expected_row_range: Optional[Range] = None): super().__init__(expected_row_count, row_id_push_down) + # The normal file is the logical anchor. A Blob column added later may + # physically cover only part of its row-id range. + self.expected_row_range = expected_row_range + self._merged_ranges: List[Range] = [] self._finished = False def add(self, file: DataFileMeta) -> None: @@ -162,18 +167,25 @@ def finish(self) -> None: if self._finished: return - merged = Range.sort_and_merge_overlap( - [blob_file.row_id_range() for blob_file in self._files], - True, - True, - ) - self._row_count = sum(row_range.count() for row_range in merged) - if self.expected_row_count >= 0 and self._row_count > self.expected_row_count: - raise ValueError( - f"Blob files row count exceed the expect {self.expected_row_count}" - ) + merged = self._merged_physical_ranges() + physical_row_count = sum(row_range.count() for row_range in merged) + if self.expected_row_range is not None: + for row_range in merged: + if (row_range.from_ < self.expected_row_range.from_ + or row_range.to > self.expected_row_range.to): + raise ValueError( + f"Blob file range {row_range} should be within normal " + f"file range {self.expected_row_range}." + ) + self._row_count = self.expected_row_range.count() + else: + self._row_count = physical_row_count + if self.expected_row_count >= 0 and self._row_count > self.expected_row_count: + raise ValueError( + f"Blob files row count exceed the expect {self.expected_row_count}" + ) - if not self.row_id_push_down: + if not self.row_id_push_down and self.expected_row_range is None: if len(merged) != 1: raise ValueError("Blob file bunch should always contain a contiguous row range.") if self.expected_row_count >= 0 and self._row_count != self.expected_row_count: @@ -181,20 +193,38 @@ def finish(self) -> None: "The merged row count of blob file bunch should be aligned " f"with normal files, expect {self.expected_row_count}, got {self._row_count}." ) + self._merged_ranges = merged self._finished = True def row_count(self) -> int: self.finish() return self._row_count + def logical_range(self) -> Range: + if self.expected_row_range is not None: + return self.expected_row_range + merged = self._merged_physical_ranges() + if not merged: + raise ValueError("Blob bunch should not be empty.") + return Range(merged[0].from_, merged[-1].to) + def sequential_read_optimize(self) -> bool: if not self._files: raise ValueError("Blob bunch should not be empty.") max_sequence_number = self._files[0].max_sequence_number - return all( + same_sequence = all( file.max_sequence_number == max_sequence_number for file in self._files ) + merged = self._merged_physical_ranges() + return (same_sequence and len(merged) == 1 + and merged[0] == self.logical_range()) + + def _merged_physical_ranges(self) -> List[Range]: + if self._finished: + return self._merged_ranges + return Range.sort_and_merge_overlap( + [blob_file.row_id_range() for blob_file in self._files], True, True) def _is_special_file(self, file_name: str) -> bool: return DataFileMeta.is_blob_file(file_name) diff --git a/paimon-python/pypaimon/read/split_read.py b/paimon-python/pypaimon/read/split_read.py index b647e713c435..cc0cca45c955 100644 --- a/paimon-python/pypaimon/read/split_read.py +++ b/paimon-python/pypaimon/read/split_read.py @@ -1319,14 +1319,14 @@ def _create_union_reader(self, need_merge_files: List[DataFileMeta], deletion_ve # Validate row counts and first row IDs (skip when row ranges are pushed down) row_count = fields_files[0].row_count() - first_row_id = fields_files[0].files()[0].first_row_id + first_row_id = self._bunch_first_row_id(fields_files[0]) if self.row_ranges is None: for bunch in fields_files: if bunch.row_count() != row_count: raise ValueError( "All files in a field merge split should have the same row count.") - if bunch.files()[0].first_row_id != first_row_id: + if self._bunch_first_row_id(bunch) != first_row_id: raise ValueError( "All files in a field merge split should have the same " "first row id and could not be null." @@ -1387,12 +1387,10 @@ def _create_union_reader(self, need_merge_files: List[DataFileMeta], deletion_ve # non-empty bunch reader created below must return the same row-id # sequence. Keep row_ranges and the group-level deletion vector # applied uniformly across normal, blob, and vector bunches. - if len(bunch.files()) == 1: - suppliers = [lambda r=self._create_file_reader( - bunch.files()[0], read_field_names, deletion_vector - ): r] - file_record_readers[i] = MergeAllBatchReader(suppliers, batch_size=batch_size) - elif DataFileMeta.is_blob_file(first_file.file_name): + is_blob_bunch = isinstance(bunch, BlobBunch) + if (is_blob_bunch + and (len(bunch.files()) != 1 + or not bunch.sequential_read_optimize())): file_reader_suppliers = [ ( file, @@ -1415,7 +1413,13 @@ def _create_union_reader(self, need_merge_files: List[DataFileMeta], deletion_ve deletion_vector=deletion_vector, batch_size=batch_size, blob_parallelism=self._blob_parallelism, + logical_ranges=[bunch.logical_range()], ) + elif len(bunch.files()) == 1: + suppliers = [lambda r=self._create_file_reader( + bunch.files()[0], read_field_names, deletion_vector + ): r] + file_record_readers[i] = MergeAllBatchReader(suppliers, batch_size=batch_size) else: # Create concatenated reader for multiple files suppliers = [ @@ -1480,13 +1484,15 @@ def _split_field_bunches(self, need_merge_files: List[DataFileMeta]) -> List[Fie blob_bunch_map = {} vector_bunch_map = {} row_count = -1 + row_range = None row_id_push_down = self.row_ranges is not None for file in need_merge_files: if DataFileMeta.is_blob_file(file.file_name): field_id = self._get_field_id_from_write_cols(file) if field_id not in blob_bunch_map: - blob_bunch_map[field_id] = BlobBunch(row_count, row_id_push_down) + blob_bunch_map[field_id] = BlobBunch( + row_count, row_id_push_down, row_range) blob_bunch_map[field_id].add(file) elif DataFileMeta.is_vector_file(file.file_name): field_id = self._get_field_id_from_write_cols(file) @@ -1496,6 +1502,7 @@ def _split_field_bunches(self, need_merge_files: List[DataFileMeta]) -> List[Fie else: fields_files.append(DataBunch(file)) row_count = file.row_count + row_range = file.row_id_range() for bunch in blob_bunch_map.values(): bunch.finish() @@ -1503,6 +1510,12 @@ def _split_field_bunches(self, need_merge_files: List[DataFileMeta]) -> List[Fie fields_files.extend(vector_bunch_map.values()) return fields_files + @staticmethod + def _bunch_first_row_id(bunch: FieldBunch) -> int: + if isinstance(bunch, BlobBunch): + return bunch.logical_range().from_ + return bunch.files()[0].first_row_id + def _get_field_id_from_write_cols(self, file: DataFileMeta) -> int: """Get field ID from write columns for blob/vector files.""" if not file.write_cols or len(file.write_cols) == 0: diff --git a/paimon-python/pypaimon/tests/blob_table_test.py b/paimon-python/pypaimon/tests/blob_table_test.py index 4b7e99a93706..c08289b4efdb 100755 --- a/paimon-python/pypaimon/tests/blob_table_test.py +++ b/paimon-python/pypaimon/tests/blob_table_test.py @@ -1779,6 +1779,89 @@ def test_blob_update_single_row_at_first_position(self): self.assertEqual(by_id[0], b'updated-0') self.assertEqual(by_id[9], b'blob-9') + def test_update_new_blob_column_writes_full_normal_range(self): + from pypaimon import Schema + from pypaimon.read.reader.format_blob_reader import FormatBlobReader + from pypaimon.schema.data_types import AtomicType + from pypaimon.write.blob_format_writer import BlobFormatWriter + + table_name = 'test_db.blob_update_new_column' + normal_schema = pa.schema([ + ('id', pa.int32()), + ('name', pa.string()), + ]) + schema = Schema.from_pyarrow_schema(normal_schema, options={ + 'row-tracking.enabled': 'true', + 'data-evolution.enabled': 'true', + }) + self.catalog.create_table(table_name, schema, False) + table = self.catalog.get_table(table_name) + + write_builder = table.new_batch_write_builder() + writer = write_builder.new_write() + writer.write_arrow(pa.Table.from_pydict({ + 'id': [1, 2, 3], + 'name': ['a', 'b', 'c'], + }, schema=normal_schema)) + write_builder.new_commit().commit(writer.prepare_commit()) + writer.close() + + self.catalog.alter_table( + table_name, + [SchemaChange.add_column('blob_data', AtomicType('BLOB'))], + False, + ) + table = self.catalog.get_table(table_name) + row_id_builder = table.new_read_builder().with_projection( + ['id', '_ROW_ID']) + row_id_result = row_id_builder.new_read().to_arrow( + row_id_builder.new_scan().plan().splits()).sort_by('id') + first_row_id = row_id_result.column('_ROW_ID')[0].as_py() + + update_builder = table.new_batch_write_builder() + table_update = update_builder.new_update().with_update_type( + ['blob_data']) + update_data = pa.Table.from_pydict({ + '_ROW_ID': pa.array([first_row_id], type=pa.int64()), + 'blob_data': pa.array([b'updated-blob'], type=pa.large_binary()), + }) + update_messages = table_update.update_by_arrow_with_row_id(update_data) + update_blob_files = [ + file + for message in update_messages + for file in message.new_files + if file.file_name.endswith('.blob') + ] + self.assertEqual(len(update_blob_files), 1) + self.assertEqual(update_blob_files[0].first_row_id, first_row_id) + self.assertEqual(update_blob_files[0].row_count, 3) + + blob_reader = FormatBlobReader( + file_io=table.file_io, + file_path=update_blob_files[0].file_path, + read_fields=['blob_data'], + full_fields=[table.field_dict['blob_data']], + push_down_predicate=None, + blob_as_descriptor=False, + ) + blob_lengths = list(blob_reader.blob_lengths) + blob_reader.close() + self.assertEqual(blob_lengths.count(BlobFormatWriter.NULL_LENGTH), 2) + self.assertNotIn(BlobFormatWriter.PLACE_HOLDER_LENGTH, blob_lengths) + + update_builder.new_commit().commit(update_messages) + read_builder = table.new_read_builder() + result = read_builder.new_read().to_arrow( + read_builder.new_scan().plan().splits()).sort_by('id') + self.assertEqual( + result.select(['id', 'blob_data']).to_pylist(), + [ + {'id': 1, 'blob_data': b'updated-blob'}, + {'id': 2, 'blob_data': None}, + {'id': 3, 'blob_data': None}, + ], + ) + def test_blob_update_all_rows_full_span(self): from pypaimon import Schema diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index a20f24dcfe9b..4c6a7b1932ae 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -44,6 +44,7 @@ from pypaimon.table.row.blob import Blob, BlobData, BlobRef, BlobDescriptor, BlobViewStruct, BlobView from pypaimon.table.row.generic_row import GenericRowDeserializer, GenericRowSerializer, GenericRow from pypaimon.table.row.row_kind import RowKind +from pypaimon.utils.range import Range class MockFileIO: @@ -580,6 +581,52 @@ def descriptor_offsets(batch): reader.close() self.assertTrue(created_by_file["third.blob"][0].closed) + def test_blob_fallback_batch_reader_fills_logical_range_gaps_with_null(self): + class FakeBlobReader: + def __init__(self): + self._file_io = None + self.file_path = "partial.blob" + self.blob_lengths = [20] + self.blob_offsets = [0] + self._input_stream = None + + def close(self): + pass + + file = DataFileMeta( + file_name="partial.blob", + file_size=0, + row_count=1, + min_key=None, + max_key=None, + key_stats=None, + value_stats=None, + min_sequence_number=1, + max_sequence_number=1, + schema_id=0, + level=0, + extra_files=[], + first_row_id=2, + file_path="partial.blob", + ) + reader = BlobFallbackBatchReader( + [(file, FakeBlobReader)], + "picture", + pa.large_binary(), + blob_as_descriptor=True, + logical_ranges=[Range(0, 4)], + ) + + batch = reader.read_arrow_batch() + + values = batch.column("picture").to_pylist() + self.assertIsNone(values[0]) + self.assertIsNone(values[1]) + self.assertEqual(4, BlobDescriptor.deserialize(values[2]).offset) + self.assertIsNone(values[3]) + self.assertIsNone(values[4]) + self.assertIsNone(reader.read_arrow_batch()) + def test_blob_fallback_batch_reader_materializes_selected_values_in_parallel(self): class RecordingFileIO: def __init__(self): diff --git a/paimon-python/pypaimon/tests/field_bunch_test.py b/paimon-python/pypaimon/tests/field_bunch_test.py index 0b6772d284eb..6cc4363e4d17 100644 --- a/paimon-python/pypaimon/tests/field_bunch_test.py +++ b/paimon-python/pypaimon/tests/field_bunch_test.py @@ -28,11 +28,13 @@ def __init__( file_name: str, first_row_id: int, row_count: int, - write_cols=None): + write_cols=None, + max_sequence_number: int = 0): self.file_name = file_name self.first_row_id = first_row_id self.row_count = row_count self.write_cols = write_cols or ["blob_col"] + self.max_sequence_number = max_sequence_number def row_id_range(self) -> Range: return Range( @@ -58,6 +60,8 @@ def test_finish_merges_ranges_once(self): bunch.finish() self.assertEqual(1, mocked_merge.call_count) self.assertEqual(1000, bunch.row_count()) + self.assertEqual(Range(0, 999), bunch.logical_range()) + self.assertTrue(bunch.sequential_read_optimize()) self.assertEqual(1, mocked_merge.call_count) def test_finish_preserves_overlapping_versions(self): @@ -125,6 +129,28 @@ def test_finish_rejects_row_count_exceeding_expected(self): with self.assertRaisesRegex(ValueError, "row count exceed"): bunch.finish() + def test_finish_uses_normal_range_for_partial_blob_coverage(self): + bunch = BlobBunch( + expected_row_count=10, + expected_row_range=Range(100, 109), + ) + bunch.add(_BlobFile("partial.blob", 103, 5)) + + bunch.finish() + + self.assertEqual(10, bunch.row_count()) + self.assertEqual(Range(100, 109), bunch.logical_range()) + + def test_finish_rejects_range_outside_normal_file(self): + bunch = BlobBunch( + expected_row_count=10, + expected_row_range=Range(100, 109), + ) + bunch.add(_BlobFile("outside.blob", 99, 2)) + + with self.assertRaisesRegex(ValueError, "within normal file range"): + bunch.finish() + if __name__ == "__main__": unittest.main() diff --git a/paimon-python/pypaimon/write/table_update_by_row_id.py b/paimon-python/pypaimon/write/table_update_by_row_id.py index 0a68c682e252..943ca29b7920 100644 --- a/paimon-python/pypaimon/write/table_update_by_row_id.py +++ b/paimon-python/pypaimon/write/table_update_by_row_id.py @@ -17,7 +17,7 @@ import bisect from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, List, Optional, Set, Tuple import numpy as np import pyarrow as pa @@ -424,30 +424,32 @@ def _merge_update_with_original( column_names: List[str], first_row_id: int, blob_object_columns: Optional[Dict[str, List[Any]]] = None, + blob_columns_with_baseline: Optional[Set[str]] = None, ) -> Tuple[Optional[pa.Table], Dict[str, List[object]]]: """Merge update data with original data, preserving row order. For rows that have updates, use the update values. For rows without updates, use the original values (if available). - Blob delta files cover ``[first_row_id, first_row_id + max_updated_pos]`` - — anchored at the original file's first_row_id, spanning up to and - including the last updated row. The span is NOT shrunk at the head: - ``BlobFallbackRecordReader`` resolves placeholders by relative offset - from the delta file's ``first_row_id``, so anchoring anywhere other - than the original ``first_row_id`` would misalign unchanged rows - before ``min_updated_pos`` with the older blob file. This anchor is - the same in every blob column being updated. + Blob delta files are anchored at the original file's first_row_id. + Existing Blob columns only need to span through their last updated + row because placeholders can fall back to older files. A Blob column + without a physical baseline spans the complete normal file and writes + real nulls for unchanged rows, so it does not create placeholders with + nowhere to fall back. Args: original_data: Original data from the file (may be None if no columns need to be read) update_data: Update data (may contain only partial rows) column_names: Column names being updated first_row_id: The first_row_id of this file group + blob_object_columns: Blob objects supplied by row-based updates + blob_columns_with_baseline: Blob columns backed by an existing physical file Returns: - Normal merged PyArrow Table, and per-blob-column values list. All - blob value lists have the same length (= ``max_updated_pos + 1``). + Normal merged PyArrow Table, and per-blob-column values list. + Blob value list lengths may differ when the updated columns do not + all have physical baselines. """ # Get the _ROW_ID values from update_data to determine which rows are being updated @@ -471,25 +473,32 @@ def _merge_update_with_original( sorted_updates = None # Caller (_write_by_first_row_id) only enters this method with a # non-empty group, so update_positions is non-empty here. - blob_row_count = max(update_positions) + 1 + blob_columns_with_baseline = blob_columns_with_baseline or set() for col_name in column_names: if self._is_blob_column(col_name): + has_baseline = col_name in blob_columns_with_baseline + blob_row_count = ( + max(update_positions) + 1 + if has_baseline else original_data.num_rows + ) + missing_value = ( + self._blob_placeholder(col_name) + if has_baseline else None + ) if blob_object_columns and col_name in blob_object_columns: update_values = blob_object_columns[col_name] - placeholder = self._blob_placeholder(col_name) blob_columns[col_name] = [ update_values[update_positions[i]] if i in update_positions - else placeholder + else missing_value for i in range(blob_row_count) ] continue update_col = update_by_col[col_name] - placeholder = self._blob_placeholder(col_name) blob_columns[col_name] = [ update_col[update_positions[i]].as_py() if i in update_positions - else placeholder + else missing_value for i in range(blob_row_count) ] continue @@ -744,12 +753,20 @@ def _write_group( writes a single output file (rolling disabled) for the group. """ original_data = self._read_original_file_data(first_row_id, column_names) + _, target_files = self._first_row_id_index[first_row_id] + blob_columns_with_baseline = { + column_name + for file in target_files + if DataFileMeta.is_blob_file(file.file_name) + for column_name in (file.write_cols or []) + } merged_data, blob_columns = self._merge_update_with_original( original_data, data, column_names, first_row_id, blob_object_columns, + blob_columns_with_baseline, ) partition_tuple = tuple(partition.values) @@ -811,13 +828,10 @@ def _write_group( def _assign_update_file_metadata(new_files: List[DataFileMeta], first_row_id: int, column_names: List[str], blob_columns: Dict[str, List[object]]): - # All blob columns share the same anchored span (see - # _merge_update_with_original docstring), so any column's length is - # the per-blob delta-file row count. - blob_row_count = ( - len(next(iter(blob_columns.values()))) if blob_columns else 0 - ) - blob_end = first_row_id + blob_row_count + blob_ends = { + column_name: first_row_id + len(values) + for column_name, values in blob_columns.items() + } blob_starts = {} # BlobWriter.prepare_commit preserves write/rolling order, which is required # for assigning continuous row-id ranges to rolled blob files. @@ -831,6 +845,7 @@ def _assign_update_file_metadata(new_files: List[DataFileMeta], first_row_id: in blob_column = file.write_cols[0] blob_start = blob_starts.get(blob_column, first_row_id) next_blob_start = blob_start + file.row_count + blob_end = blob_ends[blob_column] if next_blob_start > blob_end: raise RuntimeError( f"Blob update file {file.file_name} row-id range " @@ -846,6 +861,7 @@ def _assign_update_file_metadata(new_files: List[DataFileMeta], first_row_id: in file.first_row_id = first_row_id for blob_column, next_blob_start in blob_starts.items(): + blob_end = blob_ends[blob_column] if next_blob_start != blob_end: raise RuntimeError( f"Blob update column {blob_column} covers row ids "