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 b31acd09d6 [python] supports reading data evolution tables with
deletionVectorEnabled (#8404)
b31acd09d6 is described below
commit b31acd09d6a14a4773350c13cc24c920012f9cf2
Author: Faiz <[email protected]>
AuthorDate: Wed Jul 1 22:45:45 2026 +0800
[python] supports reading data evolution tables with deletionVectorEnabled
(#8404)
---
.../test/java/org/apache/paimon/JavaPyE2ETest.java | 132 ++++++++++
paimon-python/dev/run_mixed_tests.sh | 40 ++-
paimon-python/pypaimon/deletionvectors/__init__.py | 9 +-
.../apply_deletion_vector_reader.py | 86 +++++--
.../pypaimon/read/reader/concat_batch_reader.py | 26 +-
.../pypaimon/read/scanner/file_scanner.py | 2 +
paimon-python/pypaimon/read/split.py | 13 +-
paimon-python/pypaimon/read/split_read.py | 77 +++++-
.../tests/data_evolution_deletion_vector_test.py | 270 +++++++++++++++++++++
.../pypaimon/tests/e2e/java_py_read_write_test.py | 22 ++
.../pypaimon/tests/reader_split_generator_test.py | 19 +-
.../pypaimon/utils/data_evolution_utils.py | 51 ++++
12 files changed, 710 insertions(+), 37 deletions(-)
diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index 695cc29d76..8579fb9e85 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -24,19 +24,29 @@ import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.BinaryVector;
+import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.DataFormatTestUtil;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.variant.GenericVariant;
+import org.apache.paimon.deletionvectors.BitmapDeletionVector;
+import org.apache.paimon.deletionvectors.DeletionVector;
+import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIOFinder;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
@@ -66,6 +76,8 @@ import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RangeHelper;
import org.apache.paimon.utils.TraceableFileIO;
import org.junit.jupiter.api.BeforeEach;
@@ -79,7 +91,9 @@ import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.UUID;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -95,7 +109,9 @@ import static org.apache.paimon.CoreOptions.TARGET_FILE_SIZE;
import static org.apache.paimon.data.DataFormatTestUtil.internalRowToString;
import static
org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexOptions.BITMAP_INDEX_COMPRESSION;
import static
org.apache.paimon.globalindex.btree.BTreeIndexOptions.BTREE_INDEX_COMPRESSION;
+import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET;
import static org.apache.paimon.table.SimpleTableTestBase.getResult;
+import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
import static org.assertj.core.api.Assertions.assertThat;
/** Mixed language overwrite test for Java and Python interoperability. */
@@ -1656,6 +1672,53 @@ public class JavaPyE2ETest {
}
}
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testDataEvolutionDeletionVectorWrite() throws Exception {
+ Identifier identifier = identifier("data_evolution_dv_test");
+ catalog.dropTable(identifier, true);
+ Schema schema =
+ Schema.newBuilder()
+ .column("f0", DataTypes.INT())
+ .column("f1", DataTypes.STRING())
+ .column("f2", DataTypes.STRING())
+ .column("f3", DataTypes.BLOB())
+ .option(CoreOptions.FILE_FORMAT.key(), "parquet")
+ .option(TARGET_FILE_SIZE.key(), "128 MB")
+ .option(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "1 b")
+ .option(ROW_TRACKING_ENABLED.key(), "true")
+ .option(DATA_EVOLUTION_ENABLED.key(), "true")
+ .option(DELETION_VECTORS_ENABLED.key(), "true")
+ .build();
+ catalog.createTable(identifier, schema, false);
+
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
+ for (int batch = 0; batch < 3; batch++) {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int rowId = batch * 5; rowId < batch * 5 + 5; rowId++) {
+ write.write(
+ GenericRow.of(
+ rowId,
+ BinaryString.fromString("name-" + rowId),
+ BinaryString.fromString("base-" + rowId),
+ new BlobData(new byte[] {(byte) rowId})));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ commitDeletionVectors(
+ table,
+ Arrays.asList(
+ new E2eDvSpec(new Range(0, 4), 1, 4),
+ new E2eDvSpec(new Range(5, 9), 5, 6, 7, 8, 9),
+ new E2eDvSpec(new Range(10, 14), 10, 12)));
+
+ LOG.info("data_evolution_dv_test: written 15 rows with deletion
vectors");
+ }
+
/** Read data evolution tables written by Python. */
@Test
@EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
@@ -1676,6 +1739,75 @@ public class JavaPyE2ETest {
}
}
+ private void commitDeletionVectors(FileStoreTable table, List<E2eDvSpec>
deletionVectorSpecs)
+ throws Exception {
+ BaseAppendDeleteFileMaintainer maintainer =
+ BaseAppendDeleteFileMaintainer.forUnawareAppend(
+ table.store().newIndexFileHandler(),
+ table.latestSnapshot().get(),
+ BinaryRow.EMPTY_ROW);
+ Map<Range, String> anchorFiles = anchorFilesByRange(table);
+
+ for (E2eDvSpec spec : deletionVectorSpecs) {
+ DeletionVector deletionVector = new BitmapDeletionVector();
+ for (long rowId : spec.deletedRowIds) {
+ deletionVector.delete(rowId - spec.range.from);
+ }
+ maintainer.notifyNewDeletionVector(anchorFiles.get(spec.range),
deletionVector);
+ }
+
+ List<IndexFileMeta> newIndexFiles = new ArrayList<>();
+ List<IndexFileMeta> deletedIndexFiles = new ArrayList<>();
+ for (IndexManifestEntry entry : maintainer.persist()) {
+ if (entry.kind() == FileKind.ADD) {
+ newIndexFiles.add(entry.indexFile());
+ } else if (entry.kind() == FileKind.DELETE) {
+ deletedIndexFiles.add(entry.indexFile());
+ }
+ }
+
+ table.newBatchWriteBuilder()
+ .newCommit()
+ .commit(
+ Collections.singletonList(
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ UNAWARE_BUCKET,
+ null,
+ new DataIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ newIndexFiles,
+ deletedIndexFiles),
+ CompactIncrement.emptyIncrement())));
+ }
+
+ private Map<Range, String> anchorFilesByRange(FileStoreTable table) {
+ List<DataFileMeta> dataFiles =
+ table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .collect(Collectors.toList());
+ RangeHelper<DataFileMeta> rangeHelper = new
RangeHelper<>(DataFileMeta::nonNullRowIdRange);
+ Map<Range, String> result = new HashMap<>();
+ for (List<DataFileMeta> group :
rangeHelper.mergeOverlappingRanges(dataFiles)) {
+ DataFileMeta anchor = retrieveAnchorFile(group, file -> file);
+ result.put(anchor.nonNullRowIdRange(), anchor.fileName());
+ }
+ return result;
+ }
+
+ private static class E2eDvSpec {
+
+ private final Range range;
+ private final long[] deletedRowIds;
+
+ private E2eDvSpec(Range range, long... deletedRowIds) {
+ this.range = range;
+ this.deletedRowIds = deletedRowIds;
+ }
+ }
+
private void setFirstRowId(List<CommitMessage> messages, long firstRowId) {
messages.forEach(
c -> {
diff --git a/paimon-python/dev/run_mixed_tests.sh
b/paimon-python/dev/run_mixed_tests.sh
index efd066524c..40839700b7 100755
--- a/paimon-python/dev/run_mixed_tests.sh
+++ b/paimon-python/dev/run_mixed_tests.sh
@@ -825,6 +825,30 @@ run_data_evolution_test() {
return 0
}
+run_data_evolution_deletion_vector_test() {
+ echo -e "${YELLOW}=== Running Data Evolution Deletion Vector Test (Java
Write, Python Read) ===${NC}"
+
+ cd "$PROJECT_ROOT"
+
+ echo "Running Maven test for
JavaPyE2ETest.testDataEvolutionDeletionVectorWrite..."
+ if mvn test
-Dtest=org.apache.paimon.JavaPyE2ETest#testDataEvolutionDeletionVectorWrite -pl
paimon-core -q -Drun.e2e.tests=true; then
+ echo -e "${GREEN}✓ Java data evolution deletion vector write completed
successfully${NC}"
+ else
+ echo -e "${RED}✗ Java data evolution deletion vector write failed${NC}"
+ return 1
+ fi
+
+ cd "$PAIMON_PYTHON_DIR"
+ echo "Running Python test for
JavaPyReadWriteTest.test_read_data_evolution_deletion_vector_table..."
+ if python -m pytest
java_py_read_write_test.py::JavaPyReadWriteTest::test_read_data_evolution_deletion_vector_table
-v; then
+ echo -e "${GREEN}✓ Python data evolution deletion vector read
completed successfully${NC}"
+ return 0
+ else
+ echo -e "${RED}✗ Python data evolution deletion vector read
failed${NC}"
+ return 1
+ fi
+}
+
run_data_evolution_py_write_test() {
echo -e "${YELLOW}=== Running Data Evolution Test (Python Write, Java
Read) ===${NC}"
@@ -1009,6 +1033,7 @@ main() {
local blob_compact_conflict_result=0
local blob_alter_compact_result=0
local data_evolution_result=0
+ local data_evolution_deletion_vector_result=0
local data_evolution_py_write_result=0
local vector_dedicated_java_write_result=0
local vector_dedicated_py_write_result=0
@@ -1230,6 +1255,13 @@ main() {
echo ""
+ # Run data evolution deletion vector test (Java write, Python read)
+ if ! run_data_evolution_deletion_vector_test; then
+ data_evolution_deletion_vector_result=1
+ fi
+
+ echo ""
+
# Run data evolution test (Python write, Java read)
if ! run_data_evolution_py_write_test; then
data_evolution_py_write_result=1
@@ -1409,6 +1441,12 @@ main() {
echo -e "${RED}✗ Data Evolution Test (Java Write, Python Read):
FAILED${NC}"
fi
+ if [[ $data_evolution_deletion_vector_result -eq 0 ]]; then
+ echo -e "${GREEN}✓ Data Evolution Deletion Vector Test (Java Write,
Python Read): PASSED${NC}"
+ else
+ echo -e "${RED}✗ Data Evolution Deletion Vector Test (Java Write,
Python Read): FAILED${NC}"
+ fi
+
if [[ $data_evolution_py_write_result -eq 0 ]]; then
echo -e "${GREEN}✓ Data Evolution Test (Python Write, Java Read):
PASSED${NC}"
else
@@ -1438,7 +1476,7 @@ main() {
# Clean up warehouse directory after all tests
cleanup_warehouse
- if [[ $java_write_result -eq 0 && $python_read_result -eq 0 &&
$python_write_result -eq 0 && $java_read_result -eq 0 && $pk_dv_result -eq 0 &&
$btree_index_result -eq 0 && $btree_raw_fallback_result -eq 0 &&
$bitmap_index_result -eq 0 && $compressed_global_index_result -eq 0 &&
$compressed_text_result -eq 0 && $tantivy_fulltext_result -eq 0 &&
$lumina_vector_result -eq 0 && $lumina_vector_btree_result -eq 0 &&
$vindex_vector_result -eq 0 && $vindex_vector_raw_fallback_result -eq 0 && [...]
+ if [[ $java_write_result -eq 0 && $python_read_result -eq 0 &&
$python_write_result -eq 0 && $java_read_result -eq 0 && $pk_dv_result -eq 0 &&
$btree_index_result -eq 0 && $btree_raw_fallback_result -eq 0 &&
$bitmap_index_result -eq 0 && $compressed_global_index_result -eq 0 &&
$compressed_text_result -eq 0 && $tantivy_fulltext_result -eq 0 &&
$lumina_vector_result -eq 0 && $lumina_vector_btree_result -eq 0 &&
$vindex_vector_result -eq 0 && $vindex_vector_raw_fallback_result -eq 0 && [...]
echo -e "${GREEN}🎉 All tests passed! Java-Python interoperability
verified.${NC}"
return 0
else
diff --git a/paimon-python/pypaimon/deletionvectors/__init__.py
b/paimon-python/pypaimon/deletionvectors/__init__.py
index aa7e77dad5..b208082a46 100644
--- a/paimon-python/pypaimon/deletionvectors/__init__.py
+++ b/paimon-python/pypaimon/deletionvectors/__init__.py
@@ -17,11 +17,16 @@
from pypaimon.deletionvectors.deletion_vector import DeletionVector
from pypaimon.deletionvectors.bitmap_deletion_vector import
BitmapDeletionVector
-from pypaimon.deletionvectors.apply_deletion_vector_reader import
ApplyDeletionVectorReader, ApplyDeletionRecordIterator
+from pypaimon.deletionvectors.apply_deletion_vector_reader import (
+ ApplyDeletionVectorReader,
+ ApplyDeletionRecordIterator,
+ PositionMappedDeletionVector,
+)
__all__ = [
'DeletionVector',
'BitmapDeletionVector',
'ApplyDeletionVectorReader',
- 'ApplyDeletionRecordIterator'
+ 'ApplyDeletionRecordIterator',
+ 'PositionMappedDeletionVector',
]
diff --git
a/paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py
b/paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py
index 2642b7dc84..f491ecfa3f 100644
--- a/paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py
+++ b/paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-from typing import Optional
+from typing import List, Optional
import pyarrow
from pyarrow import RecordBatch
@@ -23,45 +23,83 @@ from pyarrow import RecordBatch
from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
from pypaimon.read.reader.iface.record_iterator import RecordIterator
from pypaimon.deletionvectors.deletion_vector import DeletionVector
-from pypaimon.utils.roaring_bitmap import RoaringBitmap
from pypaimon.read.reader.iface.record_reader import RecordReader
+class PositionMappedDeletionVector:
+ """
+ Adapts a deletion vector to the returned positions of the current reader.
+ """
+
+ def __init__(
+ self,
+ deletion_vector: DeletionVector,
+ file_offset: int = 0,
+ row_positions: Optional[List[int]] = None,
+ ):
+ self._deletion_vector = deletion_vector
+ self._file_offset = file_offset
+ self._row_positions = row_positions
+
+ def is_deleted(self, position: int) -> bool:
+ return
self._deletion_vector.is_deleted(self._mapped_position(position))
+
+ def _mapped_position(self, position: int) -> int:
+ if self._row_positions is None:
+ return self._file_offset + position
+ if position >= len(self._row_positions):
+ raise ValueError(
+ "Deletion vector row positions are fewer than returned rows."
+ )
+ return self._file_offset + self._row_positions[position]
+
+
class ApplyDeletionVectorReader(RecordBatchReader):
"""
A RecordReader which applies DeletionVector to filter records.
"""
- def __init__(self, reader: RecordReader, deletion_vector: DeletionVector):
+ def __init__(
+ self,
+ reader: RecordReader,
+ deletion_vector,
+ ):
"""
Initialize an ApplyDeletionVectorReader.
Args:
reader: The underlying record reader.
- deletion_vector: The deletion vector to apply.
+ deletion_vector: The deletion vector to apply. It should already
+ be mapped to the returned positions of the underlying reader.
"""
self._reader = reader
self._deletion_vector = deletion_vector
+ self._returned_position = 0
def reader(self) -> RecordReader:
return self._reader
- def deletion_vector(self) -> DeletionVector:
+ def deletion_vector(self):
return self._deletion_vector
def read_arrow_batch(self) -> Optional[RecordBatch]:
self._reader: RecordBatchReader
- arrow_batch = self._reader.read_arrow_batch()
- if arrow_batch is None:
- return None
- # Remove the deleted rows from the batch
- range_bitmap = RoaringBitmap()
- return_batch_pos = self._reader.return_batch_pos()
- range_bitmap.add_range(return_batch_pos - arrow_batch.num_rows,
return_batch_pos - 1)
- intersection_bitmap = RoaringBitmap.remove_all(range_bitmap,
self._deletion_vector.bit_map())
- added_row_list = [x - (return_batch_pos - arrow_batch.num_rows) for x
in
- list(intersection_bitmap)]
- return arrow_batch.take(pyarrow.array(added_row_list,
type=pyarrow.int32()))
+ # Skip physical batches that become empty after DV filtering. Some
merge readers
+ # treat a zero-row batch as EOF and would otherwise stop before later
live rows.
+ while True:
+ arrow_batch = self._reader.read_arrow_batch()
+ if arrow_batch is None:
+ return None
+
+ start = self._returned_position
+ end = start + arrow_batch.num_rows
+ self._returned_position = end
+ keep_indices = [
+ i for i, position in enumerate(range(start, end))
+ if not self._deletion_vector.is_deleted(position)
+ ]
+ if keep_indices:
+ return arrow_batch.take(pyarrow.array(keep_indices,
type=pyarrow.int32()))
def read_batch(self) -> Optional[RecordIterator]:
"""
@@ -87,7 +125,11 @@ class ApplyDeletionRecordIterator(RecordIterator):
to filter out deleted records.
"""
- def __init__(self, iterator: RecordIterator, deletion_vector:
DeletionVector):
+ def __init__(
+ self,
+ iterator: RecordIterator,
+ deletion_vector,
+ ):
"""
Initialize an ApplyDeletionRecordIterator.
@@ -97,15 +139,16 @@ class ApplyDeletionRecordIterator(RecordIterator):
"""
self._iterator = iterator
self._deletion_vector = deletion_vector
+ self._returned_position = -1
def iterator(self) -> RecordIterator:
return self._iterator
- def deletion_vector(self) -> DeletionVector:
+ def deletion_vector(self):
return self._deletion_vector
def returned_position(self) -> int:
- return self._iterator.return_pos()
+ return self._returned_position
def next(self) -> Optional[object]:
"""
@@ -123,6 +166,7 @@ class ApplyDeletionRecordIterator(RecordIterator):
if record is None:
return None
- # Check if the current position is deleted
- if not
self._deletion_vector.is_deleted(self._iterator.return_pos()):
+ self._returned_position += 1
+ position = self._returned_position
+ if not self._deletion_vector.is_deleted(position):
return record
diff --git a/paimon-python/pypaimon/read/reader/concat_batch_reader.py
b/paimon-python/pypaimon/read/reader/concat_batch_reader.py
index 67d1a7c40a..51b6f573a6 100644
--- a/paimon-python/pypaimon/read/reader/concat_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/concat_batch_reader.py
@@ -126,6 +126,7 @@ class MergeAllBatchReader(RecordBatchReader):
)
else:
self.merged_batch = None
+ return None
dataset = ds.InMemoryDataset(self.merged_batch)
self.reader = dataset.scanner(batch_size=self._batch_size).to_reader()
return self.reader.read_next_batch()
@@ -240,12 +241,17 @@ 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):
+ blob_as_descriptor: bool = False, deletion_vector=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._blob_as_descriptor = blob_as_descriptor
+ if deletion_vector is None:
+ self._deletion_vector_range = None
+ self._deletion_vector = None
+ else:
+ self._deletion_vector_range, self._deletion_vector =
deletion_vector
self._returned = False
self._readers: List[RecordBatchReader] = []
@@ -319,7 +325,11 @@ class BlobFallbackBatchReader(RecordBatchReader):
]
if self._row_ranges is not None:
ranges = Range.and_(ranges, self._row_ranges)
- return self._expand_ranges(ranges)
+ return [
+ row_id
+ for row_id in self._expand_ranges(ranges)
+ if not self._is_deleted(row_id)
+ ]
def _selected_row_ids(self, file: DataFileMeta) -> List[int]:
ranges = [file.row_id_range()]
@@ -335,6 +345,18 @@ class BlobFallbackBatchReader(RecordBatchReader):
for row_id in range(row_range.from_, row_range.to + 1)
]
+ def _is_deleted(self, row_id: int) -> bool:
+ if self._deletion_vector is None:
+ return False
+ if not self._deletion_vector_range.contains(row_id):
+ raise ValueError(
+ f"Deletion vector range {self._deletion_vector_range} "
+ f"should contain blob row id {row_id}."
+ )
+ return self._deletion_vector.is_deleted(
+ row_id - self._deletion_vector_range.from_
+ )
+
def _read_blob_values(self, file: DataFileMeta, supplier: Callable) ->
List[object]:
reader = supplier()
if reader is None:
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index fe46abd67b..d6c7ad6f36 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -565,6 +565,8 @@ class FileScanner:
"""
if self.limit is None:
return splits
+ if self.data_evolution and self.deletion_vectors_enabled:
+ return splits
if self._has_non_partition_filter():
return splits
diff --git a/paimon-python/pypaimon/read/split.py
b/paimon-python/pypaimon/read/split.py
index 46513656ae..c318a15b2c 100644
--- a/paimon-python/pypaimon/read/split.py
+++ b/paimon-python/pypaimon/read/split.py
@@ -184,6 +184,10 @@ class DataSplit(Split):
for file in self._files:
if file.first_row_id is None:
return False
+ if self.data_deletion_files is not None:
+ for deletion_file in self.data_deletion_files:
+ if deletion_file is not None and deletion_file.cardinality is
None:
+ return False
return True
def _data_evolution_merged_row_count(self) -> int:
@@ -198,4 +202,11 @@ class DataSplit(Split):
return 0
ranges = Range.sort_and_merge_overlap(file_ranges, True, True)
- return sum([r.count() for r in ranges])
+ row_count = sum([r.count() for r in ranges])
+ if self.data_deletion_files is not None:
+ row_count -= sum(
+ deletion_file.cardinality
+ for deletion_file in self.data_deletion_files
+ if deletion_file is not None
+ )
+ return row_count
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index 231b525bbf..fce5378956 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -23,7 +23,10 @@ from typing import Callable, Dict, List, Optional, Tuple
from pypaimon.common.merge_engine_dispatch import build_merge_function
from pypaimon.common.options.core_options import CoreOptions, MergeEngine
from pypaimon.common.predicate import Predicate
-from pypaimon.deletionvectors import ApplyDeletionVectorReader
+from pypaimon.deletionvectors import (
+ ApplyDeletionVectorReader,
+ PositionMappedDeletionVector,
+)
from pypaimon.deletionvectors.deletion_vector import DeletionVector
from pypaimon.globalindex import Range
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
@@ -68,6 +71,7 @@ from pypaimon.read.sliced_split import SlicedSplit
from pypaimon.schema.data_types import DataField, PyarrowFieldParser
from pypaimon.table.special_fields import SpecialFields
from pypaimon.globalindex.indexed_split import IndexedSplit
+from pypaimon.utils.data_evolution_utils import retrieve_anchor_file
KEY_PREFIX = "_KEY_"
KEY_FIELD_ID_START = 1000000
@@ -1009,19 +1013,22 @@ class DataEvolutionSplitRead(SplitRead):
"""Core read logic: split_by_row_id -> suppliers -> ConcatBatchReader
-> filter."""
files = self.split.files
suppliers = []
+ self._genarate_deletion_file_readers()
# Split files by row ID
split_by_row_id = self._split_by_row_id(files)
for need_merge_files in split_by_row_id:
+ deletion_vector = self._read_deletion_vector(need_merge_files)
if len(need_merge_files) == 1 or not self.read_fields:
# No need to merge fields, just create a single file reader
suppliers.append(
- lambda f=need_merge_files[0]: self._create_file_reader(f,
self._get_final_read_data_fields())
+ lambda f=need_merge_files[0], dv=deletion_vector:
self._create_file_reader(
+ f, self._get_final_read_data_fields(), dv)
)
else:
suppliers.append(
- lambda files=need_merge_files:
self._create_union_reader(files)
+ lambda files=need_merge_files, dv=deletion_vector:
self._create_union_reader(files, dv)
)
merge_reader = ConcatBatchReader(
@@ -1043,6 +1050,51 @@ class DataEvolutionSplitRead(SplitRead):
return reader
+ def _read_deletion_vector(self, need_merge_files: List[DataFileMeta]):
+ if not getattr(self, "deletion_file_readers", None):
+ return None
+
+ anchor = retrieve_anchor_file(need_merge_files)
+ dv_factory = self.deletion_file_readers.get(anchor.file_name)
+ if dv_factory is None:
+ return None
+
+ deletion_vector = dv_factory()
+ if deletion_vector is None:
+ return None
+
+ return anchor.row_id_range(), deletion_vector
+
+ def _apply_deletion_vector(self, reader, reader_range: Range,
deletion_vector):
+ if reader is None or deletion_vector is None:
+ return reader
+
+ dv_range, dv = deletion_vector
+ if dv.is_empty():
+ return reader
+
+ if dv_range.from_ > reader_range.from_ or dv_range.to <
reader_range.to:
+ raise ValueError(
+ f"Deletion vector range {dv_range} should contain reader range
{reader_range}."
+ )
+
+ mapped_dv = PositionMappedDeletionVector(
+ dv,
+ reader_range.from_ - dv_range.from_,
+ self._selected_local_positions(reader_range),
+ )
+ return ApplyDeletionVectorReader(reader, mapped_dv)
+
+ def _selected_local_positions(self, reader_range: Range) ->
Optional[List[int]]:
+ if self.row_ranges is None:
+ return None
+ selected = Range.and_([reader_range], self.row_ranges)
+ return [
+ row_id - reader_range.from_
+ for row_range in selected
+ for row_id in range(row_range.from_, row_range.to + 1)
+ ]
+
def _create_prescan_reader(self, field_names):
"""Create a prescan reader by constructing a new DataEvolutionSplitRead
instance that only projects the specified field names.
@@ -1116,7 +1168,7 @@ class DataEvolutionSplitRead(SplitRead):
return split_by_row_id
- def _create_union_reader(self, need_merge_files: List[DataFileMeta]) ->
RecordReader:
+ def _create_union_reader(self, need_merge_files: List[DataFileMeta],
deletion_vector=None) -> RecordReader:
"""Create a DataEvolutionFileReader for merging multiple files."""
# Split field bunches
fields_files = self._split_field_bunches(need_merge_files)
@@ -1187,10 +1239,13 @@ class DataEvolutionSplitRead(SplitRead):
table_fields = self.read_fields
self.read_fields = read_fields # create reader based on
read_fields
batch_size = self.table.options.read_batch_size()
- # Create reader for this bunch
+ # DataEvolutionMergeReader aligns fields by row ordinal, so
every
+ # 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
+ 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):
@@ -1213,12 +1268,14 @@ class DataEvolutionSplitRead(SplitRead):
).field(0).type,
self.row_ranges,
CoreOptions.blob_as_descriptor(self.table.options),
+ deletion_vector=deletion_vector,
)
else:
# Create concatenated reader for multiple files
suppliers = [
partial(self._create_file_reader, file=file,
- read_fields=read_field_names) for file in
bunch.files()
+ read_fields=read_field_names,
+ deletion_vector=deletion_vector) for file in
bunch.files()
]
file_record_readers[i] = MergeAllBatchReader(suppliers,
batch_size=batch_size)
self.read_fields = table_fields
@@ -1232,14 +1289,16 @@ class DataEvolutionSplitRead(SplitRead):
output_schema = PyarrowFieldParser.from_paimon_schema(all_read_fields)
return DataEvolutionMergeReader(row_offsets, field_offsets,
file_record_readers, schema=output_schema)
- def _create_file_reader(self, file: DataFileMeta, read_fields: [str]) ->
Optional[RecordReader]:
+ def _create_file_reader(
+ self, file: DataFileMeta, read_fields: [str],
deletion_vector=None) -> Optional[RecordReader]:
"""Create a file reader for a single file."""
- return self.file_reader_supplier(
+ reader = self.file_reader_supplier(
file=file,
for_merge_read=False,
read_fields=read_fields,
row_tracking_enabled=True,
row_ranges=self.row_ranges)
+ return self._apply_deletion_vector(reader, file.row_id_range(),
deletion_vector)
def _create_raw_blob_file_reader(
self, file: DataFileMeta, read_fields: [str]) ->
Optional[FormatBlobReader]:
diff --git
a/paimon-python/pypaimon/tests/data_evolution_deletion_vector_test.py
b/paimon-python/pypaimon/tests/data_evolution_deletion_vector_test.py
new file mode 100644
index 0000000000..650adb95f2
--- /dev/null
+++ b/paimon-python/pypaimon/tests/data_evolution_deletion_vector_test.py
@@ -0,0 +1,270 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import unittest
+
+import pyarrow as pa
+
+from pypaimon.deletionvectors.apply_deletion_vector_reader import (
+ ApplyDeletionVectorReader,
+ PositionMappedDeletionVector,
+)
+from pypaimon.deletionvectors.bitmap_deletion_vector import
BitmapDeletionVector
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.read.reader.concat_batch_reader import (
+ BlobFallbackBatchReader,
+ DataEvolutionMergeReader,
+ MergeAllBatchReader,
+)
+from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+from pypaimon.read.split import DataSplit
+from pypaimon.table.row.blob import Blob, BlobData
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.source.deletion_file import DeletionFile
+from pypaimon.utils.range import Range
+from pypaimon.utils.data_evolution_utils import retrieve_anchor_file
+
+
+class _OneBatchReader(RecordBatchReader):
+ def __init__(self, values):
+ self._batch = pa.record_batch([pa.array(values, type=pa.int64())],
names=["v"])
+ self._returned = False
+
+ def read_arrow_batch(self):
+ if self._returned:
+ return None
+ self._returned = True
+ return self._batch
+
+ def close(self):
+ pass
+
+
+class _BlobFallbackBatchReaderForTest(BlobFallbackBatchReader):
+ def __init__(
+ self, files, values_by_file_name, row_ranges=None, deletion_vector=None
+ ):
+ super().__init__(
+ [(file, lambda: None) for file in files],
+ "blob_col",
+ pa.binary(),
+ row_ranges=row_ranges,
+ blob_as_descriptor=False,
+ deletion_vector=deletion_vector,
+ )
+ self._values_by_file_name = values_by_file_name
+
+ def _read_blob_values(self, file, supplier):
+ values = self._values_by_file_name[file.file_name]
+ return [
+ values[row_id - file.first_row_id]
+ for row_id in self._selected_row_ids(file)
+ ]
+
+
+def _file(name, first_row_id, row_count, max_sequence_number):
+ empty_row = GenericRow([], [])
+ return DataFileMeta(
+ file_name=name,
+ file_size=1,
+ row_count=row_count,
+ min_key=empty_row,
+ max_key=empty_row,
+ key_stats=SimpleStats.empty_stats(),
+ value_stats=SimpleStats.empty_stats(),
+ min_sequence_number=max_sequence_number,
+ max_sequence_number=max_sequence_number,
+ schema_id=0,
+ level=0,
+ extra_files=[],
+ first_row_id=first_row_id,
+ )
+
+
+class DataEvolutionDeletionVectorTest(unittest.TestCase):
+ def test_retrieve_anchor_file_uses_oldest_normal_file(self):
+ files = [
+ _file("field-2.blob", 0, 5, 1),
+ _file("normal-b.parquet", 0, 5, 1),
+ _file("normal-a.parquet", 0, 5, 1),
+ _file("newer.parquet", 0, 5, 2),
+ ]
+
+ self.assertEqual("normal-a.parquet",
retrieve_anchor_file(files).file_name)
+
+ def test_data_evolution_merged_row_count_subtracts_deletion_vectors(self):
+ split = DataSplit(
+ files=[
+ _file("anchor-0.parquet", 0, 5, 1),
+ _file("blob-0.blob", 0, 5, 2),
+ _file("anchor-5.parquet", 5, 5, 3),
+ ],
+ partition=GenericRow([], []),
+ bucket=0,
+ raw_convertible=False,
+ data_deletion_files=[
+ DeletionFile("dv", 0, 1, cardinality=2),
+ None,
+ DeletionFile("dv", 1, 1, cardinality=1),
+ ],
+ )
+
+ self.assertEqual(7, split.merged_row_count())
+
+ def test_data_evolution_merged_row_count_unknown_without_cardinality(self):
+ split = DataSplit(
+ files=[_file("anchor.parquet", 0, 5, 1)],
+ partition=GenericRow([], []),
+ bucket=0,
+ raw_convertible=False,
+ data_deletion_files=[DeletionFile("dv", 0, 1, cardinality=None)],
+ )
+
+ self.assertIsNone(split.merged_row_count())
+
+ def test_apply_deletion_vector_reader_uses_mapped_deletion_vector(self):
+ deletion_vector = BitmapDeletionVector()
+ deletion_vector.delete(12)
+ mapped_dv = PositionMappedDeletionVector(
+ deletion_vector,
+ file_offset=10,
+ row_positions=[0, 2, 4],
+ )
+
+ reader = ApplyDeletionVectorReader(
+ _OneBatchReader([0, 2, 4]),
+ mapped_dv,
+ )
+
+ batch = reader.read_arrow_batch()
+ self.assertEqual([0, 4], batch.column(0).to_pylist())
+ self.assertTrue(reader.deletion_vector().is_deleted(1))
+ self.assertFalse(reader.deletion_vector().is_deleted(2))
+
+ def test_data_evolution_merge_reader_handles_fully_deleted_file(self):
+ deletion_vector = BitmapDeletionVector()
+ deletion_vector.delete(0)
+ deletion_vector.delete(1)
+
+ field_reader = MergeAllBatchReader([
+ lambda: ApplyDeletionVectorReader(
+ _OneBatchReader([0, 1]),
+ deletion_vector,
+ )
+ ])
+ reader = DataEvolutionMergeReader(
+ row_offsets=[0],
+ field_offsets=[0],
+ readers=[field_reader],
+ schema=pa.schema([pa.field("v", pa.int64())]),
+ )
+
+ self.assertIsNone(reader.read_arrow_batch())
+
+ def test_blob_fallback_batch_reader_applies_deletion_vector(self):
+ files = [
+ _file("blob-old.blob", 0, 5, 1),
+ _file("blob-new.blob", 0, 5, 2),
+ ]
+ deletion_vector = BitmapDeletionVector()
+ deletion_vector.delete(1)
+ deletion_vector.delete(4)
+
+ reader = _BlobFallbackBatchReaderForTest(
+ files,
+ {
+ "blob-old.blob": [
+ BlobData(b"old-0"),
+ BlobData(b"old-1"),
+ BlobData(b"old-2"),
+ BlobData(b"old-3"),
+ BlobData(b"old-4"),
+ ],
+ "blob-new.blob": [
+ Blob.PLACE_HOLDER,
+ BlobData(b"new-1"),
+ BlobData(b"new-2"),
+ Blob.PLACE_HOLDER,
+ BlobData(b"new-4"),
+ ],
+ },
+ deletion_vector=(Range(0, 4), deletion_vector),
+ )
+
+ batch = reader.read_arrow_batch()
+ self.assertEqual(
+ [b"old-0", b"new-2", b"old-3"],
+ batch.column(0).to_pylist(),
+ )
+ self.assertIsNone(reader.read_arrow_batch())
+
+ def
test_data_evolution_merge_reader_aligns_blob_with_row_ranges_and_dv(self):
+ row_ranges = [Range(1, 4)]
+ deletion_vector = BitmapDeletionVector()
+ deletion_vector.delete(2)
+ deletion_vector.delete(4)
+
+ normal_reader = ApplyDeletionVectorReader(
+ _OneBatchReader([1, 2, 3, 4]),
+ PositionMappedDeletionVector(
+ deletion_vector,
+ file_offset=0,
+ row_positions=[1, 2, 3, 4],
+ ),
+ )
+ blob_reader = _BlobFallbackBatchReaderForTest(
+ [
+ _file("blob-old.blob", 0, 6, 1),
+ _file("blob-new.blob", 0, 6, 2),
+ ],
+ {
+ "blob-old.blob": [
+ BlobData(b"old-0"),
+ BlobData(b"old-1"),
+ BlobData(b"old-2"),
+ BlobData(b"old-3"),
+ BlobData(b"old-4"),
+ BlobData(b"old-5"),
+ ],
+ "blob-new.blob": [
+ Blob.PLACE_HOLDER,
+ Blob.PLACE_HOLDER,
+ BlobData(b"new-2"),
+ BlobData(b"new-3"),
+ BlobData(b"new-4"),
+ Blob.PLACE_HOLDER,
+ ],
+ },
+ row_ranges=row_ranges,
+ deletion_vector=(Range(0, 5), deletion_vector),
+ )
+ reader = DataEvolutionMergeReader(
+ row_offsets=[0, 1],
+ field_offsets=[0, 0],
+ readers=[normal_reader, blob_reader],
+ schema=pa.schema([pa.field("id", pa.int64()), pa.field("blob_col",
pa.binary())]),
+ )
+
+ batch = reader.read_arrow_batch()
+ self.assertEqual([1, 3], batch.column(0).to_pylist())
+ self.assertEqual([b"old-1", b"new-3"], batch.column(1).to_pylist())
+ self.assertIsNone(reader.read_arrow_batch())
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index ad66dd26f1..aad6f5133a 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -1473,6 +1473,28 @@ class JavaPyReadWriteTest(unittest.TestCase):
self.assertEqual(result.column('f1')[i].as_py(), f'a{i}')
self.assertEqual(result.column('f2')[i].as_py(), f'b{i}')
+ def test_read_data_evolution_deletion_vector_table(self):
+ """Read a data evolution table with deletion vectors and blob files
written by Java."""
+ table = self.catalog.get_table('default.data_evolution_dv_test')
+ read_builder = table.new_read_builder()
+ table_scan = read_builder.new_scan()
+ table_read = read_builder.new_read()
+ result = table_read.to_arrow(table_scan.plan().splits())
+ result = table_sort_by(result, 'f0')
+
+ expected_ids = [0, 2, 3, 11, 13, 14]
+ self.assertEqual(result.num_rows, len(expected_ids))
+ self.assertEqual(result.column('f0').to_pylist(), expected_ids)
+ self.assertEqual(
+ result.column('f1').to_pylist(),
+ [f'name-{i}' for i in expected_ids])
+ self.assertEqual(
+ result.column('f2').to_pylist(),
+ [f'base-{i}' for i in expected_ids])
+ self.assertEqual(
+ result.column('f3').to_pylist(),
+ [bytes([i]) for i in expected_ids])
+
@parameterized.expand(get_file_format_params())
def test_py_write_data_evolution_table(self, file_format):
"""Python writes data evolution tables for Java to read."""
diff --git a/paimon-python/pypaimon/tests/reader_split_generator_test.py
b/paimon-python/pypaimon/tests/reader_split_generator_test.py
index b29d88dc23..32ae1ddbf2 100644
--- a/paimon-python/pypaimon/tests/reader_split_generator_test.py
+++ b/paimon-python/pypaimon/tests/reader_split_generator_test.py
@@ -333,7 +333,12 @@ class ApplyPushDownLimitUnitTest(unittest.TestCase):
"""Mock-driven coverage of ``FileScanner._apply_push_down_limit``."""
@staticmethod
- def _apply(splits, limit, has_non_partition_filter=False):
+ def _apply(
+ splits,
+ limit,
+ has_non_partition_filter=False,
+ data_evolution=False,
+ deletion_vectors_enabled=False):
from pypaimon.read.scanner.file_scanner import FileScanner
class _FakeScanner:
@@ -341,6 +346,8 @@ class ApplyPushDownLimitUnitTest(unittest.TestCase):
scanner = _FakeScanner()
scanner.limit = limit
+ scanner.data_evolution = data_evolution
+ scanner.deletion_vectors_enabled = deletion_vectors_enabled
scanner._has_non_partition_filter = lambda: has_non_partition_filter
return FileScanner._apply_push_down_limit(scanner, splits)
@@ -399,6 +406,16 @@ class ApplyPushDownLimitUnitTest(unittest.TestCase):
[s_raw, s_raw, s_raw], limit=5, has_non_partition_filter=True)
self.assertEqual(len(result), 3)
+ def test_data_evolution_deletion_vectors_disable_limit_pushdown(self):
+ s_raw = self._split(raw_convertible=True, row_count=10,
merged_row_count=10)
+ result = self._apply(
+ [s_raw, s_raw, s_raw],
+ limit=5,
+ data_evolution=True,
+ deletion_vectors_enabled=True,
+ )
+ self.assertEqual(len(result), 3)
+
if __name__ == '__main__':
unittest.main()
diff --git a/paimon-python/pypaimon/utils/data_evolution_utils.py
b/paimon-python/pypaimon/utils/data_evolution_utils.py
new file mode 100644
index 0000000000..06c5d36265
--- /dev/null
+++ b/paimon-python/pypaimon/utils/data_evolution_utils.py
@@ -0,0 +1,51 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Utilities for data-evolution tables."""
+
+from typing import Callable, Iterable, TypeVar
+
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+
+T = TypeVar("T")
+
+
+def retrieve_anchor_file(
+ entries: Iterable[T],
+ file_meta_func: Callable[[T], DataFileMeta] = lambda entry: entry,
+) -> T:
+ """Return the oldest normal file in a data-evolution row-range group."""
+ anchor = None
+ anchor_key = None
+
+ for entry in entries:
+ meta = file_meta_func(entry)
+ if DataFileMeta.is_blob_file(meta.file_name) or
DataFileMeta.is_vector_file(meta.file_name):
+ continue
+
+ key = (meta.max_sequence_number, meta.file_name)
+ if anchor_key is None or key < anchor_key:
+ anchor = entry
+ anchor_key = key
+
+ if anchor is None:
+ raise ValueError(
+ "Data-evolution deletion vectors should have a normal anchor file "
+ "in each row range group."
+ )
+
+ return anchor
