This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 37c1bea125 [core][python] Fix partial ranges when updating newly
added Blob columns (#9388)
37c1bea125 is described below
commit 37c1bea1253809cd3b6eb600ad61a29a65977278
Author: zhoulii <[email protected]>
AuthorDate: Wed Aug 26 09:47:34 2026 +0800
[core][python] Fix partial ranges when updating newly added Blob columns
(#9388)
---
.../paimon/operation/BlobFallbackRecordReader.java | 51 +++++++++++--
.../paimon/operation/DataEvolutionSplitRead.java | 65 ++++++++++++-----
.../operation/BlobFallbackRecordReaderTest.java | 44 ++++++++++++
.../paimon/operation/DataEvolutionReadTest.java | 23 +++++-
.../pypaimon/read/reader/concat_batch_reader.py | 30 +++++---
paimon-python/pypaimon/read/reader/field_bunch.py | 58 +++++++++++----
paimon-python/pypaimon/read/split_read.py | 31 +++++---
paimon-python/pypaimon/tests/blob_table_test.py | 83 ++++++++++++++++++++++
paimon-python/pypaimon/tests/blob_test.py | 47 ++++++++++++
paimon-python/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 db501c1615..927933fce0 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<InternalRow> {
RowType readRowType,
int blobIndex)
throws IOException {
+ this(
+ files,
+ readerFactory,
+ readerWrapper,
+ enclosingRange(files),
+ rowRanges,
+ readRowType,
+ blobIndex);
+ }
+
+ BlobFallbackRecordReader(
+ List<DataFileMeta> files,
+ BlobFileReaderFactory readerFactory,
+ ReaderWrapper readerWrapper,
+ Range logicalRange,
+ List<Range> 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> {
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<Long, List<DataFileMeta>> 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<InternalRow> {
}
}
+ private static Range enclosingRange(List<DataFileMeta> 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<InternalRow> 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 e064736419..2a606ee677 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 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
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 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
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 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
// 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 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
deletionVector),
(reader, range) ->
applyDeletionVector(reader, range, rowRanges,
deletionVector),
+ blobBunch.logicalRange(),
rowRanges,
readRowType,
blobIndex);
@@ -1008,14 +1010,15 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
Map<Integer, BlobFileBunch> blobBunchMap = new HashMap<>();
Map<VectorStoreBunchKey, VectorFileBunch> 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 class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
// 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 class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
List<DataFileMeta> 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 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
final List<DataFileMeta> files;
final List<Range> 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 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
@Override
public long rowCount() {
List<Range> 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<Range> 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<Range> 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 class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
}
}
- 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 859e8bc5b8..328aaa3061 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 class BlobFallbackRecordReaderTest {
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<Range> 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 24d2994957..bdb9fcc935 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.stats.SimpleStats;
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 @@ public class DataEvolutionReadTest {
@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 @@ public class DataEvolutionReadTest {
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 831e41012a..39b7dac1d6 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 @@ class BlobFallbackBatchReader(RecordBatchReader):
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 @@ class BlobFallbackBatchReader(RecordBatchReader):
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 @@ class BlobFallbackBatchReader(RecordBatchReader):
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 5474bf855a..65ab1af046 100644
--- a/paimon-python/pypaimon/read/reader/field_bunch.py
+++ b/paimon-python/pypaimon/read/reader/field_bunch.py
@@ -22,7 +22,7 @@ These classes help organize DataFileMeta objects into groups
based on their fiel
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 @@ class _SpecialFieldBunch(FieldBunch):
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 @@ class BlobBunch(_SpecialFieldBunch):
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 @@ class BlobBunch(_SpecialFieldBunch):
"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 b647e713c4..cc0cca45c9 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -1319,14 +1319,14 @@ class DataEvolutionSplitRead(SplitRead):
# 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 @@ class DataEvolutionSplitRead(SplitRead):
# 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 @@ class DataEvolutionSplitRead(SplitRead):
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 @@ class DataEvolutionSplitRead(SplitRead):
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 @@ class DataEvolutionSplitRead(SplitRead):
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 @@ class DataEvolutionSplitRead(SplitRead):
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 4b7e99a937..c08289b4ef 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/paimon-python/pypaimon/tests/blob_table_test.py
@@ -1779,6 +1779,89 @@ class DedicatedFormatWriterTest(unittest.TestCase):
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 a20f24dcfe..4c6a7b1932 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -44,6 +44,7 @@ from pypaimon.schema.data_types import ArrayType, AtomicType,
DataField, MapType
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 @@ class BlobTest(unittest.TestCase):
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 0b6772d284..6cc4363e4d 100644
--- a/paimon-python/pypaimon/tests/field_bunch_test.py
+++ b/paimon-python/pypaimon/tests/field_bunch_test.py
@@ -28,11 +28,13 @@ class _BlobFile:
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 @@ class BlobBunchTest(unittest.TestCase):
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 @@ class BlobBunchTest(unittest.TestCase):
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 0a68c682e2..943ca29b79 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 @@ class TableUpdateByRowId:
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 @@ class TableUpdateByRowId:
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 @@ class TableUpdateByRowId:
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 @@ class TableUpdateByRowId:
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 @@ class TableUpdateByRowId:
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 @@ class TableUpdateByRowId:
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 "
