You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "discivigour (via GitHub)" <gi...@apache.org> on 2025/12/07 08:48:03 UTC
[PR] [Python] Support read deletion vector pk table [paimon]
discivigour opened a new pull request, #6766:
URL: https://github.com/apache/paimon/pull/6766
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
Support read deletion vector pk table
### Tests
- python: test_pk_dv_read
- java: testPKDeletionVectorWrite
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python] Support read deletion vector pk table [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on code in PR #6766:
URL: https://github.com/apache/paimon/pull/6766#discussion_r2600833757
##########
paimon-python/dev/run_mixed_tests.sh:
##########
@@ -240,6 +268,12 @@ main() {
echo -e "${RED}✗ Java Read Test (Parquet + Lance): FAILED${NC}"
fi
+ if [[ $pk_dv_result -eq 0 ]]; then
Review Comment:
ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python] Support read deletion vector pk table [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on code in PR #6766:
URL: https://github.com/apache/paimon/pull/6766#discussion_r2597196385
##########
paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java:
##########
@@ -198,6 +220,139 @@ public void testJavaWriteReadPkTable() throws Exception {
"6, Beef, Meat, 8.0");
}
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testPKDeletionVectorWrite() throws Exception {
+ Consumer<Options> optionsSetter =
+ options -> {
+ // let level has many files
+ options.set(TARGET_FILE_SIZE, new MemorySize(1));
+ options.set(DELETION_VECTORS_ENABLED, true);
+ };
+ String tableName = "test_pk_dv";
+ Path tablePath = new Path(warehouse.toString() + "/default.db/" + tableName);
+ FileStoreTable table = createFileStoreTable(optionsSetter, tablePath);
+ StreamTableWrite write = table.newWrite(commitUser);
+ IOManager ioManager = IOManager.create(tablePath.toString());
+ write.withIOManager(ioManager);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ write.write(createRow3Cols(1, 10, 100L));
+ write.write(createRow3Cols(2, 20, 200L));
+ write.write(createRow3Cols(1, 11, 101L));
+ commit.commit(0, write.prepareCommit(true, 0));
+
+ write.write(createRow3Cols(1, 10, 1000L));
+ write.write(createRow3Cols(2, 21, 201L));
+ write.write(createRow3Cols(2, 21, 2001L));
+ commit.commit(1, write.prepareCommit(true, 1));
+
+ write.write(createRow3Cols(1, 11, 1001L));
+ write.write(createRow3Cols(2, 21, 20001L));
+ write.write(createRow3Cols(2, 22, 202L));
+ write.write(createRow3ColsWithKind(RowKind.DELETE, 1, 11, 1001L));
+ write.write(createRow3ColsWithKind(RowKind.DELETE, 2, 20, 200L));
+ commit.commit(2, write.prepareCommit(true, 2));
+
+ // test result
+ Function<InternalRow, String> rowDataToString =
+ row ->
+ internalRowToString(
+ row,
+ DataTypes.ROW(
+ DataTypes.INT(), DataTypes.INT(), DataTypes.BIGINT()));
+ List<String> result =
+ getResult(table.newRead(), table.newScan().plan().splits(), rowDataToString);
+ assertThat(result)
+ .containsExactlyInAnyOrder("+I[1, 10, 1000]", "+I[2, 21, 20001]", "+I[2, 22, 202]");
+ }
+
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testPKDeletionVectorWriteMultiBatch() throws Exception {
+ Consumer<Options> optionsSetter =
+ options -> {
+ // let level has many files
+ options.set(TARGET_FILE_SIZE, new MemorySize(128 * 1024));
+ options.set(DELETION_VECTORS_ENABLED, true);
+ };
+ String tableName = "test_pk_dv_multi_batch";
+ Path tablePath = new Path(warehouse.toString() + "/default.db/" + tableName);
+ FileStoreTable table = createFileStoreTable(optionsSetter, tablePath);
+ StreamTableWrite write = table.newWrite(commitUser);
+ IOManager ioManager = IOManager.create(tablePath.toString());
+ write.withIOManager(ioManager);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ // Write 10000 records
+ for (int i = 1; i <= 10000; i++) {
+ write.write(createRow3Cols(1, i * 10, (long) i * 100));
+ }
+ commit.commit(0, write.prepareCommit(false, 0));
+
+ // Delete the 81930th record
+ write.write(createRow3ColsWithKind(RowKind.DELETE, 1, 81930, 819300L));
+ commit.commit(1, write.prepareCommit(true, 1));
+
+ Function<InternalRow, String> rowDataToString =
+ row ->
+ internalRowToString(
+ row,
+ DataTypes.ROW(
+ DataTypes.INT(), DataTypes.INT(), DataTypes.BIGINT()));
+ List<String> result =
+ getResult(table.newRead(), table.newScan().plan().splits(), rowDataToString);
+
+ // Verify the count is 9999
+ assertThat(result).hasSize(9999);
+
+ assertThat(result).doesNotContain("+I[1, 81930, 819300L]");
+
+ assertThat(result).contains("+I[1, 10, 100]");
+ assertThat(result).contains("+I[1, 100000, 1000000]");
+ }
+
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testPKDeletionVectorWriteMultiBatchRawConvertable() throws Exception {
+ Consumer<Options> optionsSetter =
+ options -> {
+ options.set(DELETION_VECTORS_ENABLED, true);
+ };
+ String tableName = "test_pk_dv_raw_convertable";
+ Path tablePath = new Path(warehouse.toString() + "/default.db/" + tableName);
+ FileStoreTable table = createFileStoreTable(optionsSetter, tablePath);
+ StreamTableWrite write = table.newWrite(commitUser);
+ IOManager ioManager = IOManager.create(tablePath.toString());
+ write.withIOManager(ioManager);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ for (int i = 1; i <= 10000; i++) {
+ write.write(createRow3Cols(1, i * 10, (long) i * 100));
+ }
+ commit.commit(0, write.prepareCommit(false, 0));
+
+ write.write(createRow3ColsWithKind(RowKind.DELETE, 1, 81930, 819300L));
+ commit.commit(1, write.prepareCommit(true, 1));
+
+ Function<InternalRow, String> rowDataToString =
+ row ->
+ internalRowToString(
+ row,
+ DataTypes.ROW(
+ DataTypes.INT(), DataTypes.INT(), DataTypes.BIGINT()));
+ List<String> result =
+ getResult(table.newRead(), table.newScan().plan().splits(), rowDataToString);
+
+ assertThat(result).hasSize(9999);
+
+ assertThat(result).doesNotContain("+I[1, 81930, 819300L]");
+
Review Comment:
remove "L" here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python] Support read deletion vector pk table [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #6766:
URL: https://github.com/apache/paimon/pull/6766#discussion_r2596962254
##########
paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py:
##########
@@ -0,0 +1,129 @@
+# 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.
+
+from typing import Optional
+
+import pyarrow
+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 pyroaring import BitMap
+
+from pypaimon.read.reader.iface.record_reader import RecordReader
+
+
+class ApplyDeletionVectorReader(RecordBatchReader):
+ """
+ A RecordReader which applies DeletionVector to filter records.
+ """
+
+ def __init__(self, reader: RecordReader, deletion_vector: DeletionVector):
+ """
+ Initialize an ApplyDeletionVectorReader.
+
+ Args:
+ reader: The underlying record reader.
+ deletion_vector: The deletion vector to apply.
+ """
+ self._reader = reader
+ self._deletion_vector = deletion_vector
+
+ def reader(self) -> RecordReader:
+ return self._reader
+
+ def deletion_vector(self) -> DeletionVector:
+ 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 = BitMap(
+ range(self._reader.return_batch_pos() - arrow_batch.num_rows, self._reader.return_batch_pos()))
+ intersection_bitmap = range_bitmap - self._deletion_vector.bit_map()
+ added_row_list = [x - (self._reader.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()))
+
Review Comment:
It is not easy to get an empty deletion vector, so I think it is OK without empty check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python] Support read deletion vector pk table [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on PR #6766:
URL: https://github.com/apache/paimon/pull/6766#issuecomment-3630647364
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python] Support read deletion vector pk table [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on code in PR #6766:
URL: https://github.com/apache/paimon/pull/6766#discussion_r2597043796
##########
paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java:
##########
@@ -198,6 +220,139 @@ public void testJavaWriteReadPkTable() throws Exception {
"6, Beef, Meat, 8.0");
}
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testPKDeletionVectorWrite() throws Exception {
+ Consumer<Options> optionsSetter =
+ options -> {
+ // let level has many files
+ options.set(TARGET_FILE_SIZE, new MemorySize(1));
+ options.set(DELETION_VECTORS_ENABLED, true);
+ };
+ String tableName = "test_pk_dv";
+ Path tablePath = new Path(warehouse.toString() + "/default.db/" + tableName);
+ FileStoreTable table = createFileStoreTable(optionsSetter, tablePath);
+ StreamTableWrite write = table.newWrite(commitUser);
+ IOManager ioManager = IOManager.create(tablePath.toString());
+ write.withIOManager(ioManager);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ write.write(createRow3Cols(1, 10, 100L));
+ write.write(createRow3Cols(2, 20, 200L));
+ write.write(createRow3Cols(1, 11, 101L));
+ commit.commit(0, write.prepareCommit(true, 0));
+
+ write.write(createRow3Cols(1, 10, 1000L));
+ write.write(createRow3Cols(2, 21, 201L));
+ write.write(createRow3Cols(2, 21, 2001L));
+ commit.commit(1, write.prepareCommit(true, 1));
+
+ write.write(createRow3Cols(1, 11, 1001L));
+ write.write(createRow3Cols(2, 21, 20001L));
+ write.write(createRow3Cols(2, 22, 202L));
+ write.write(createRow3ColsWithKind(RowKind.DELETE, 1, 11, 1001L));
+ write.write(createRow3ColsWithKind(RowKind.DELETE, 2, 20, 200L));
+ commit.commit(2, write.prepareCommit(true, 2));
+
Review Comment:
Can you add one more commit with DElELE to make sure position is correct in each commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python] Support read deletion vector pk table [paimon]
Posted by "kaori-seasons (via GitHub)" <gi...@apache.org>.
kaori-seasons commented on code in PR #6766:
URL: https://github.com/apache/paimon/pull/6766#discussion_r2601236460
##########
paimon-python/pypaimon/deletionvectors/bitmap_deletion_vector.py:
##########
@@ -0,0 +1,165 @@
+# 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.
+
+from pypaimon.deletionvectors.deletion_vector import DeletionVector
+import struct
+import zlib
+from pyroaring import BitMap
+
+
+class BitmapDeletionVector(DeletionVector):
+ """
+ A DeletionVector based on RoaringBitmap, it only supports files with row count
+ not exceeding 2147483647 (max value for 32-bit integer).
+ """
+
+ MAGIC_NUMBER = 1581511376
+ MAGIC_NUMBER_SIZE_BYTES = 4
+ MAX_VALUE = 2147483647
+
+ def __init__(self, bitmap: BitMap = None):
+ """
+ Initialize a BitmapDeletionVector.
+
+ Args:
+ bitmap: Optional RoaringBitmap instance. If None, creates an empty bitmap.
+ """
+ self._bitmap = bitmap if bitmap is not None else BitMap()
+
+ def delete(self, position: int) -> None:
+ """
+ Marks the row at the specified position as deleted.
+
+ Args:
+ position: The position of the row to be marked as deleted.
+ """
+ self._check_position(position)
+ self._bitmap.add(position)
+
+ def is_deleted(self, position: int) -> bool:
+ """
+ Checks if the row at the specified position is deleted.
+
+ Args:
+ position: The position of the row to check.
+
+ Returns:
+ True if the row is deleted, False otherwise.
+ """
+ self._check_position(position)
+ return position in self._bitmap
+
+ def is_empty(self) -> bool:
+ return len(self._bitmap) == 0
+
+ def get_cardinality(self) -> int:
+ """
+ Returns the number of distinct integers added to the DeletionVector.
+
+ Returns:
+ The number of deleted positions.
+ """
+ return len(self._bitmap)
+
+ def merge(self, deletion_vector: DeletionVector) -> None:
+ """
+ Merge another DeletionVector to this current one.
+
+ Args:
+ deletion_vector: The other DeletionVector to merge.
+ """
+ if isinstance(deletion_vector, BitmapDeletionVector):
+ self._bitmap |= deletion_vector._bitmap
+ else:
+ raise RuntimeError("Only instance with the same class type can be merged.")
+
+ def serialize(self) -> bytes:
+ """
+ Serializes the deletion vector to bytes.
+
+ Returns:
+ The serialized bytes.
+ """
+ # Serialize the bitmap
+ bitmap_bytes = self._bitmap.serialize()
+
+ # Create the full data with magic number
+ magic_bytes = struct.pack('>I', self.MAGIC_NUMBER)
+ data = magic_bytes + bitmap_bytes
+
+ # Calculate size and checksum
+ size = len(data)
+ checksum = self._calculate_checksum(data)
+
+ # Pack: size (4 bytes) + data + checksum (4 bytes)
+ result = struct.pack('>I', size) + data + struct.pack('>I', checksum)
+
+ return result
+
+ @staticmethod
+ def deserialize_from_bytes(data: bytes) -> 'BitmapDeletionVector':
+ """
+ Deserializes a BitmapDeletionVector from bytes.
+
+ Args:
+ data: The serialized bytes (without magic number).
+
+ Returns:
+ A BitmapDeletionVector instance.
+ """
+ bitmap = BitMap.deserialize(data)
+ return BitmapDeletionVector(bitmap)
+
+ def bit_map(self):
+ return self._bitmap
+
+ def _check_position(self, position: int) -> None:
+ """
+ Checks if the position is valid.
+
+ Args:
+ position: The position to check.
+
+ Raises:
+ ValueError: If the position exceeds the maximum value.
+ """
+ if position > self.MAX_VALUE:
+ raise ValueError(
+ f"The file has too many rows, RoaringBitmap32 only supports files "
+ f"with row count not exceeding {self.MAX_VALUE}."
+ )
+
+ @staticmethod
+ def _calculate_checksum(data: bytes) -> int:
+ """
+ Calculates CRC32 checksum for the given data.
+
+ Args:
+ data: The data to calculate checksum for.
+
+ Returns:
+ The CRC32 checksum as an unsigned 32-bit integer.
+ """
+ return zlib.crc32(data) & 0xffffffff
+
+ def __eq__(self, other):
+ if not isinstance(other, BitmapDeletionVector):
+ return False
+ return self._bitmap == other._bitmap
+
+ def __hash__(self):
+ return hash(tuple(self._bitmap))
Review Comment:
BitmapDeletionVector.hash: Returns hash(tuple(self._bitmap)). This is inefficient on large bitmaps as it iterates through all bits, and it's rarely used. Consider removing the hash function or replacing it with one based on CRC/hash of serialized bytes (if a hash is truly needed).
##########
paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py:
##########
@@ -0,0 +1,129 @@
+# 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.
+
+from typing import Optional
+
+import pyarrow
+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 pyroaring import BitMap
+
+from pypaimon.read.reader.iface.record_reader import RecordReader
+
+
+class ApplyDeletionVectorReader(RecordBatchReader):
+ """
+ A RecordReader which applies DeletionVector to filter records.
+ """
+
+ def __init__(self, reader: RecordReader, deletion_vector: DeletionVector):
+ """
+ Initialize an ApplyDeletionVectorReader.
+
+ Args:
+ reader: The underlying record reader.
+ deletion_vector: The deletion vector to apply.
+ """
+ self._reader = reader
+ self._deletion_vector = deletion_vector
+
+ def reader(self) -> RecordReader:
+ return self._reader
+
+ def deletion_vector(self) -> DeletionVector:
+ 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 = BitMap(
Review Comment:
`apply_deletion_vector_reader.ApplyDeletionVectorReader.read_arrow_batch` uses `BitMap(range(batch_start, batch_end))` and then performs a set difference operation to obtain the non-deleted positions.
When the batch is very large (or the number of batches is very large), creating such a large range BitMap consumes a significant amount of memory/CPU. However, the actual deletion bitmap is usually sparse, so iterating through the deletion bitmap should be prioritized.
Impact: Performance degradation and OOM risk, especially in large tables or long intervals.
Suggested fix: Instead of generating a BitMap representing the entire batch, directly filter the positions within the current batch range from the deletion vector's bitmap (which is usually sparse and more efficient).
`pyroaring.BitMap` is iterable, so it can be iterated bit by bit and filtered.
##########
paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py:
##########
@@ -0,0 +1,129 @@
+# 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.
+
+from typing import Optional
+
+import pyarrow
+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 pyroaring import BitMap
+
+from pypaimon.read.reader.iface.record_reader import RecordReader
+
+
+class ApplyDeletionVectorReader(RecordBatchReader):
+ """
+ A RecordReader which applies DeletionVector to filter records.
+ """
+
+ def __init__(self, reader: RecordReader, deletion_vector: DeletionVector):
+ """
+ Initialize an ApplyDeletionVectorReader.
+
+ Args:
+ reader: The underlying record reader.
+ deletion_vector: The deletion vector to apply.
+ """
+ self._reader = reader
+ self._deletion_vector = deletion_vector
+
+ def reader(self) -> RecordReader:
+ return self._reader
+
+ def deletion_vector(self) -> DeletionVector:
+ 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 = BitMap(
+ range(self._reader.return_batch_pos() - arrow_batch.num_rows, self._reader.return_batch_pos()))
+ intersection_bitmap = range_bitmap - self._deletion_vector.bit_map()
+ added_row_list = [x - (self._reader.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()))
+
+ def read_batch(self) -> Optional[RecordIterator]:
+ """
+ Reads one batch with deletion vector applied.
+
+ Returns:
+ A RecordIterator with deletion filtering, or None if no more data.
+ """
+ batch = self._reader.read_batch()
+
+ if batch is None:
+ return None
+
+ return ApplyDeletionRecordIterator(batch, self._deletion_vector)
+
+ def close(self):
+ self._reader.close()
+
+
+class ApplyDeletionRecordIterator(RecordIterator):
Review Comment:
Problem: The `ApplyDeletionRecordIterator`, `RowPositionRecordIterator`, and `KeyValueWrapIterator` classes are inconsistent in how they return the position (`return_pos`): some classes use the name `returned_position`, while others use `return_pos`.
This can easily lead to `TypeError` / `AttributeError` at runtime, and makes it difficult to guarantee that wrappers (such as `ApplyDeletionVectorReader`, `RowPositionReader`) will work correctly in all scenarios.
Suggestion: Implement compatibility handling in all wrappers/adapters for cases where the object might be a Python iterator or a custom `RecordIterator`. The best practice is to require `RecordIterator` to implement `next` (to achieve fundamental consistency), but if it's not possible to modify all implementations at once, use a compatibility layer (checking `hasattr(next)` / `hasattr(next)`, etc.) to avoid crashes.
Examples:
```
# pypaimon/read/reader/iface/record_iterator.py
from abc import ABC, abstractmethod
from typing import Generic, Optional, TypeVar, Iterator
T = TypeVar("T")
class RecordIterator(Iterator[T], Generic[T], ABC):
"""
Record iterator: must implement Python iterator protocol and provide return_pos().
Implementations should:
- implement __next__() returning Optional[T] (or raise StopIteration)
- implement return_pos() returning current position index for the last returned element
"""
def __iter__(self):
return self
@abstractmethod
def __next__(self) -> T:
"""Return next record or raise StopIteration"""
...
# backward-compat helper for code that used next()
def next(self) -> Optional[T]:
try:
return self.__next__()
except StopIteration:
return None
@abstractmethod
def return_pos(self) -> int:
"""Return current position of the iterator (index of the last returned element)."""
...
```
```
class RowPositionRecordIterator(RecordIterator[tuple]):
def __init__(self):
self.reused_iterator: Optional[Iterator[tuple]] = None
self.pos = -1
def __next__(self) -> tuple:
if self.reused_iterator is None:
raise StopIteration
row_tuple = next(self.reused_iterator, None)
if row_tuple is None:
raise StopIteration
self.pos += 1
return row_tuple
def next(self) -> Optional[tuple]:
try:
return self.__next__()
except StopIteration:
return None
def return_pos(self) -> int:
return self.pos
```
Then update all custom implementations (e.g., RowPositionRecordIterator, InternalRowWrapperIterator, etc.):
Implement `next`, and maintain the position counter `self.pos` within `next` (the logic of the original `next` method is moved into the new `next`), and retain `next()` as a compatibility alias (as above).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python] Support read deletion vector pk table [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #6766:
URL: https://github.com/apache/paimon/pull/6766#discussion_r2596957050
##########
paimon-python/pypaimon/deletionvectors/deletion_vector.py:
##########
@@ -0,0 +1,142 @@
+# 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.
+
+from abc import ABC, abstractmethod
+
+from pypaimon.common.file_io import FileIO
+from pypaimon.table.source.deletion_file import DeletionFile
+
+
+class DeletionVector(ABC):
+ """
+ The DeletionVector can efficiently record the positions of rows that are deleted in a file,
+ which can then be used to filter out deleted rows when processing the file.
+ """
+
+ @abstractmethod
+ def bit_map(self):
+ """
+ Returns the bitmap of the DeletionVector.
+ """
+ pass
+
+ @abstractmethod
+ def delete(self, position: int) -> None:
+ """
+ Marks the row at the specified position as deleted.
+
+ Args:
+ position: The position of the row to be marked as deleted.
+ """
+ pass
+
+ @abstractmethod
+ def is_deleted(self, position: int) -> bool:
+ """
+ Checks if the row at the specified position is deleted.
+
+ Args:
+ position: The position of the row to check.
+
+ Returns:
+ True if the row is deleted, False otherwise.
+ """
+ pass
+
+ @abstractmethod
+ def is_empty(self) -> bool:
+ """
+ Determines if the deletion vector is empty, indicating no deletions.
+
+ Returns:
+ True if the deletion vector is empty, False if it contains deletions.
+ """
+ pass
+
+ @abstractmethod
+ def get_cardinality(self) -> int:
+ """
+ Returns the number of distinct integers added to the DeletionVector.
+
+ Returns:
+ The number of deleted positions.
+ """
+ pass
+
+ @abstractmethod
+ def merge(self, deletion_vector: 'DeletionVector') -> None:
+ """
+ Merge another DeletionVector to this current one.
+
+ Args:
+ deletion_vector: The other DeletionVector to merge.
+ """
+ pass
+
+ def checked_delete(self, position: int) -> bool:
+ """
+ Marks the row at the specified position as deleted.
+
+ Args:
+ position: The position of the row to be marked as deleted.
+
+ Returns:
+ True if the added position wasn't already deleted. False otherwise.
+ """
+ if self.is_deleted(position):
+ return False
+ else:
+ self.delete(position)
+ return True
+
+ @staticmethod
+ def read(file_io: FileIO, deletion_file: DeletionFile) -> 'DeletionVector':
+ """
+ Read a DeletionVector from a file.
+ """
+ from pypaimon.deletionvectors.bitmap_deletion_vector import BitmapDeletionVector
+
+ with file_io.new_input_stream(deletion_file.dv_index_path) as f:
+ f.seek(deletion_file.offset)
+
+ # Read bitmap length
+ bitmap_length_bytes = f.read(4)
+ bitmap_length = int.from_bytes(bitmap_length_bytes, byteorder='big')
+
+ # Read magic number
+ magic_number_bytes = f.read(4)
+ magic_number = int.from_bytes(magic_number_bytes, byteorder='big')
+
+ if magic_number == BitmapDeletionVector.MAGIC_NUMBER:
+ if deletion_file.length is not None and bitmap_length != deletion_file.length:
+ raise RuntimeError(
+ f"Size not match, actual size: {bitmap_length}, expected size: {deletion_file.length}"
+ )
+
+ # Magic number has been read, read remaining bytes
+ remaining_bytes = bitmap_length - BitmapDeletionVector.MAGIC_NUMBER_SIZE_BYTES
+ data = f.read(remaining_bytes)
+
+ # Skip CRC (4 bytes)
+ f.read(4)
+
+ return BitmapDeletionVector.deserialize_from_bytes(data)
+ else:
+ raise RuntimeError(
+ f"Invalid magic number: {magic_number}, "
+ f"expected: {BitmapDeletionVector.MAGIC_NUMBER}"
+ )
Review Comment:
Not now, it is just for iceberg compatibility.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python] Support read deletion vector pk table [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on code in PR #6766:
URL: https://github.com/apache/paimon/pull/6766#discussion_r2596945097
##########
paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py:
##########
@@ -0,0 +1,129 @@
+# 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.
+
+from typing import Optional
+
+import pyarrow
+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 pyroaring import BitMap
+
+from pypaimon.read.reader.iface.record_reader import RecordReader
+
+
+class ApplyDeletionVectorReader(RecordBatchReader):
+ """
+ A RecordReader which applies DeletionVector to filter records.
+ """
+
+ def __init__(self, reader: RecordReader, deletion_vector: DeletionVector):
+ """
+ Initialize an ApplyDeletionVectorReader.
+
+ Args:
+ reader: The underlying record reader.
+ deletion_vector: The deletion vector to apply.
+ """
+ self._reader = reader
+ self._deletion_vector = deletion_vector
+
+ def reader(self) -> RecordReader:
+ return self._reader
+
+ def deletion_vector(self) -> DeletionVector:
+ 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 = BitMap(
+ range(self._reader.return_batch_pos() - arrow_batch.num_rows, self._reader.return_batch_pos()))
+ intersection_bitmap = range_bitmap - self._deletion_vector.bit_map()
+ added_row_list = [x - (self._reader.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()))
+
Review Comment:
We can add empty check for self._deletion_vector, if empty, then return arrow_batch directly
##########
paimon-python/pypaimon/read/split_read.py:
##########
@@ -281,17 +285,33 @@ def _construct_partition_mapping(self) -> List[int]:
return mapping
+ def _genarate_deletion_file_readers(self):
Review Comment:
typo: genarate -> generate
##########
paimon-python/pypaimon/deletionvectors/deletion_vector.py:
##########
@@ -0,0 +1,142 @@
+# 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.
+
+from abc import ABC, abstractmethod
+
+from pypaimon.common.file_io import FileIO
+from pypaimon.table.source.deletion_file import DeletionFile
+
+
+class DeletionVector(ABC):
+ """
+ The DeletionVector can efficiently record the positions of rows that are deleted in a file,
+ which can then be used to filter out deleted rows when processing the file.
+ """
+
+ @abstractmethod
+ def bit_map(self):
+ """
+ Returns the bitmap of the DeletionVector.
+ """
+ pass
+
+ @abstractmethod
+ def delete(self, position: int) -> None:
+ """
+ Marks the row at the specified position as deleted.
+
+ Args:
+ position: The position of the row to be marked as deleted.
+ """
+ pass
+
+ @abstractmethod
+ def is_deleted(self, position: int) -> bool:
+ """
+ Checks if the row at the specified position is deleted.
+
+ Args:
+ position: The position of the row to check.
+
+ Returns:
+ True if the row is deleted, False otherwise.
+ """
+ pass
+
+ @abstractmethod
+ def is_empty(self) -> bool:
+ """
+ Determines if the deletion vector is empty, indicating no deletions.
+
+ Returns:
+ True if the deletion vector is empty, False if it contains deletions.
+ """
+ pass
+
+ @abstractmethod
+ def get_cardinality(self) -> int:
+ """
+ Returns the number of distinct integers added to the DeletionVector.
+
+ Returns:
+ The number of deleted positions.
+ """
+ pass
+
+ @abstractmethod
+ def merge(self, deletion_vector: 'DeletionVector') -> None:
+ """
+ Merge another DeletionVector to this current one.
+
+ Args:
+ deletion_vector: The other DeletionVector to merge.
+ """
+ pass
+
+ def checked_delete(self, position: int) -> bool:
+ """
+ Marks the row at the specified position as deleted.
+
+ Args:
+ position: The position of the row to be marked as deleted.
+
+ Returns:
+ True if the added position wasn't already deleted. False otherwise.
+ """
+ if self.is_deleted(position):
+ return False
+ else:
+ self.delete(position)
+ return True
+
+ @staticmethod
+ def read(file_io: FileIO, deletion_file: DeletionFile) -> 'DeletionVector':
+ """
+ Read a DeletionVector from a file.
+ """
+ from pypaimon.deletionvectors.bitmap_deletion_vector import BitmapDeletionVector
+
+ with file_io.new_input_stream(deletion_file.dv_index_path) as f:
+ f.seek(deletion_file.offset)
+
+ # Read bitmap length
+ bitmap_length_bytes = f.read(4)
+ bitmap_length = int.from_bytes(bitmap_length_bytes, byteorder='big')
+
+ # Read magic number
+ magic_number_bytes = f.read(4)
+ magic_number = int.from_bytes(magic_number_bytes, byteorder='big')
+
+ if magic_number == BitmapDeletionVector.MAGIC_NUMBER:
+ if deletion_file.length is not None and bitmap_length != deletion_file.length:
+ raise RuntimeError(
+ f"Size not match, actual size: {bitmap_length}, expected size: {deletion_file.length}"
+ )
+
+ # Magic number has been read, read remaining bytes
+ remaining_bytes = bitmap_length - BitmapDeletionVector.MAGIC_NUMBER_SIZE_BYTES
+ data = f.read(remaining_bytes)
+
+ # Skip CRC (4 bytes)
+ f.read(4)
+
+ return BitmapDeletionVector.deserialize_from_bytes(data)
+ else:
+ raise RuntimeError(
+ f"Invalid magic number: {magic_number}, "
+ f"expected: {BitmapDeletionVector.MAGIC_NUMBER}"
+ )
Review Comment:
Do we need support 64-bit bitmap?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python] Support read deletion vector pk table [paimon]
Posted by "kaori-seasons (via GitHub)" <gi...@apache.org>.
kaori-seasons commented on code in PR #6766:
URL: https://github.com/apache/paimon/pull/6766#discussion_r2601258956
##########
paimon-python/pypaimon/manifest/index_manifest_file.py:
##########
@@ -0,0 +1,95 @@
+# 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.
+
+from io import BytesIO
+from typing import List
+
+import fastavro
+
+from pypaimon.index.deletion_vector_meta import DeletionVectorMeta
+from pypaimon.index.index_file_meta import IndexFileMeta
+from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+from pypaimon.table.row.generic_row import GenericRowDeserializer
+
+
+class IndexManifestFile:
+ """Index manifest file reader for reading index manifest entries."""
+
+ DELETION_VECTORS_INDEX = "DELETION_VECTORS"
+
+ def __init__(self, table):
+ from pypaimon.table.file_store_table import FileStoreTable
+
+ self.table: FileStoreTable = table
+ manifest_path = table.table_path.rstrip('/')
Review Comment:
Problem: In `_to_deletion_files`, the `index_path` is constructed directly using `self.table.table_path + '/index'`, without considering `index_file.external_path` (the `IndexFileMeta` object has an `external_path` field).
Impact: When parsing `dv_meta_record` using fastavro, hardcoded field names like `dv_meta_record['f0']`, 'f1', 'f2', etc., are used. If the Avro schema field names are different, an error will occur.
Suggestion: When constructing `dv_file_path`, if `index_file.external_path` exists, it should be used preferentially.
When parsing `dv_meta_record`, use a defensive approach with key lookup (`get`) and be compatible with multiple naming conventions (or use schema-driven parsing with the fastavro reader).
Examples:
```
def _to_deletion_files(self, index_entry) -> Dict[str, DeletionFile]:
deletion_files = {}
index_file = index_entry.index_file
if not index_file.dv_ranges:
return deletion_files
# prefer external_path when present
index_path = index_file.external_path if index_file.external_path else self.table.table_path.rstrip('/') + '/index'
dv_file_path = f"{index_path}/{index_file.file_name}"
for key, dv_meta in index_file.dv_ranges.items():
# dv_meta is expected to be DeletionVectorMeta
deletion_file = DeletionFile(
dv_index_path=dv_file_path,
offset=dv_meta.offset,
length=dv_meta.length,
cardinality=dv_meta.cardinality
)
deletion_files[key] = deletion_file
return deletion_files
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python] Support read deletion vector pk table [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on code in PR #6766:
URL: https://github.com/apache/paimon/pull/6766#discussion_r2597039260
##########
paimon-python/pypaimon/deletionvectors/deletion_vector.py:
##########
@@ -0,0 +1,142 @@
+# 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.
+
+from abc import ABC, abstractmethod
+
+from pypaimon.common.file_io import FileIO
+from pypaimon.table.source.deletion_file import DeletionFile
+
+
+class DeletionVector(ABC):
+ """
+ The DeletionVector can efficiently record the positions of rows that are deleted in a file,
+ which can then be used to filter out deleted rows when processing the file.
+ """
+
+ @abstractmethod
+ def bit_map(self):
+ """
+ Returns the bitmap of the DeletionVector.
+ """
+ pass
+
+ @abstractmethod
+ def delete(self, position: int) -> None:
+ """
+ Marks the row at the specified position as deleted.
+
+ Args:
+ position: The position of the row to be marked as deleted.
+ """
+ pass
+
+ @abstractmethod
+ def is_deleted(self, position: int) -> bool:
+ """
+ Checks if the row at the specified position is deleted.
+
+ Args:
+ position: The position of the row to check.
+
+ Returns:
+ True if the row is deleted, False otherwise.
+ """
+ pass
+
+ @abstractmethod
+ def is_empty(self) -> bool:
+ """
+ Determines if the deletion vector is empty, indicating no deletions.
+
+ Returns:
+ True if the deletion vector is empty, False if it contains deletions.
+ """
+ pass
+
+ @abstractmethod
+ def get_cardinality(self) -> int:
+ """
+ Returns the number of distinct integers added to the DeletionVector.
+
+ Returns:
+ The number of deleted positions.
+ """
+ pass
+
+ @abstractmethod
+ def merge(self, deletion_vector: 'DeletionVector') -> None:
+ """
+ Merge another DeletionVector to this current one.
+
+ Args:
+ deletion_vector: The other DeletionVector to merge.
+ """
+ pass
+
+ def checked_delete(self, position: int) -> bool:
+ """
+ Marks the row at the specified position as deleted.
+
+ Args:
+ position: The position of the row to be marked as deleted.
+
+ Returns:
+ True if the added position wasn't already deleted. False otherwise.
+ """
+ if self.is_deleted(position):
+ return False
+ else:
+ self.delete(position)
+ return True
+
+ @staticmethod
+ def read(file_io: FileIO, deletion_file: DeletionFile) -> 'DeletionVector':
+ """
+ Read a DeletionVector from a file.
+ """
+ from pypaimon.deletionvectors.bitmap_deletion_vector import BitmapDeletionVector
+
+ with file_io.new_input_stream(deletion_file.dv_index_path) as f:
+ f.seek(deletion_file.offset)
+
+ # Read bitmap length
+ bitmap_length_bytes = f.read(4)
+ bitmap_length = int.from_bytes(bitmap_length_bytes, byteorder='big')
+
+ # Read magic number
+ magic_number_bytes = f.read(4)
+ magic_number = int.from_bytes(magic_number_bytes, byteorder='big')
+
+ if magic_number == BitmapDeletionVector.MAGIC_NUMBER:
+ if deletion_file.length is not None and bitmap_length != deletion_file.length:
+ raise RuntimeError(
+ f"Size not match, actual size: {bitmap_length}, expected size: {deletion_file.length}"
+ )
+
+ # Magic number has been read, read remaining bytes
+ remaining_bytes = bitmap_length - BitmapDeletionVector.MAGIC_NUMBER_SIZE_BYTES
+ data = f.read(remaining_bytes)
+
+ # Skip CRC (4 bytes)
+ f.read(4)
+
+ return BitmapDeletionVector.deserialize_from_bytes(data)
+ else:
+ raise RuntimeError(
+ f"Invalid magic number: {magic_number}, "
+ f"expected: {BitmapDeletionVector.MAGIC_NUMBER}"
+ )
Review Comment:
> Not now, it is just for iceberg compatibility.
👌
##########
paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py:
##########
@@ -0,0 +1,129 @@
+# 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.
+
+from typing import Optional
+
+import pyarrow
+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 pyroaring import BitMap
+
+from pypaimon.read.reader.iface.record_reader import RecordReader
+
+
+class ApplyDeletionVectorReader(RecordBatchReader):
+ """
+ A RecordReader which applies DeletionVector to filter records.
+ """
+
+ def __init__(self, reader: RecordReader, deletion_vector: DeletionVector):
+ """
+ Initialize an ApplyDeletionVectorReader.
+
+ Args:
+ reader: The underlying record reader.
+ deletion_vector: The deletion vector to apply.
+ """
+ self._reader = reader
+ self._deletion_vector = deletion_vector
+
+ def reader(self) -> RecordReader:
+ return self._reader
+
+ def deletion_vector(self) -> DeletionVector:
+ 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 = BitMap(
+ range(self._reader.return_batch_pos() - arrow_batch.num_rows, self._reader.return_batch_pos()))
+ intersection_bitmap = range_bitmap - self._deletion_vector.bit_map()
+ added_row_list = [x - (self._reader.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()))
+
Review Comment:
> It is not easy to get an empty deletion vector, so I think it is OK without empty check.
👌
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python] Support read deletion vector pk table [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on code in PR #6766:
URL: https://github.com/apache/paimon/pull/6766#discussion_r2597054632
##########
paimon-python/dev/run_mixed_tests.sh:
##########
@@ -240,6 +268,12 @@ main() {
echo -e "${RED}✗ Java Read Test (Parquet + Lance): FAILED${NC}"
fi
+ if [[ $pk_dv_result -eq 0 ]]; then
Review Comment:
We should add pk_dv_result return value check in line 282 too
##########
paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java:
##########
@@ -198,6 +220,139 @@ public void testJavaWriteReadPkTable() throws Exception {
"6, Beef, Meat, 8.0");
}
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testPKDeletionVectorWrite() throws Exception {
+ Consumer<Options> optionsSetter =
+ options -> {
+ // let level has many files
+ options.set(TARGET_FILE_SIZE, new MemorySize(1));
+ options.set(DELETION_VECTORS_ENABLED, true);
+ };
+ String tableName = "test_pk_dv";
+ Path tablePath = new Path(warehouse.toString() + "/default.db/" + tableName);
+ FileStoreTable table = createFileStoreTable(optionsSetter, tablePath);
+ StreamTableWrite write = table.newWrite(commitUser);
+ IOManager ioManager = IOManager.create(tablePath.toString());
+ write.withIOManager(ioManager);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ write.write(createRow3Cols(1, 10, 100L));
+ write.write(createRow3Cols(2, 20, 200L));
+ write.write(createRow3Cols(1, 11, 101L));
+ commit.commit(0, write.prepareCommit(true, 0));
+
+ write.write(createRow3Cols(1, 10, 1000L));
+ write.write(createRow3Cols(2, 21, 201L));
+ write.write(createRow3Cols(2, 21, 2001L));
+ commit.commit(1, write.prepareCommit(true, 1));
+
+ write.write(createRow3Cols(1, 11, 1001L));
+ write.write(createRow3Cols(2, 21, 20001L));
+ write.write(createRow3Cols(2, 22, 202L));
+ write.write(createRow3ColsWithKind(RowKind.DELETE, 1, 11, 1001L));
+ write.write(createRow3ColsWithKind(RowKind.DELETE, 2, 20, 200L));
+ commit.commit(2, write.prepareCommit(true, 2));
+
Review Comment:
Can you add one more commit with DElELE to make sure position is correct in each commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python] Support read deletion vector pk table [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #6766:
URL: https://github.com/apache/paimon/pull/6766
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org