You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/06/19 12:31:58 UTC
(paimon) branch master updated: [python] Support reading bitmap global index (#8289)
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 76e32fc68f [python] Support reading bitmap global index (#8289)
76e32fc68f is described below
commit 76e32fc68f97c976de2808794fda26174383d39e
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri Jun 19 20:31:52 2026 +0800
[python] Support reading bitmap global index (#8289)
Add Python-side support for reading Java-written bitmap global index
files and cover the Java/Python interoperability path.
---
.../test/java/org/apache/paimon/JavaPyE2ETest.java | 179 ++++++++++
paimon-python/dev/run_mixed_tests.sh | 76 ++++-
.../pypaimon/common/options/core_options.py | 28 ++
paimon-python/pypaimon/globalindex/__init__.py | 10 +
.../pypaimon/globalindex/bitmap/__init__.py | 26 ++
.../globalindex/bitmap/bitmap_index_reader.py | 361 +++++++++++++++++++++
.../bitmap/lazy_filtered_bitmap_reader.py | 34 ++
.../pypaimon/globalindex/block_compression.py | 319 ++++++++++++++++++
.../pypaimon/globalindex/btree/block_reader.py | 2 +-
.../globalindex/btree/btree_file_meta_selector.py | 115 +------
.../pypaimon/globalindex/btree/btree_index_meta.py | 45 +--
.../globalindex/btree/btree_index_reader.py | 16 +-
.../pypaimon/globalindex/btree/key_serializer.py | 144 +-------
.../btree/lazy_filtered_btree_reader.py | 198 +----------
.../globalindex/btree/memory_slice_input.py | 145 +--------
.../pypaimon/globalindex/btree/sst_file_reader.py | 57 +---
.../pypaimon/globalindex/global_index_evaluator.py | 2 +
.../pypaimon/globalindex/global_index_reader.py | 5 +
.../pypaimon/globalindex/global_index_scanner.py | 34 +-
.../globalindex/{btree => }/key_serializer.py | 45 +--
.../globalindex/{btree => }/memory_slice_input.py | 95 ++----
.../globalindex/offset_global_index_reader.py | 5 +
.../globalindex/sorted_file_global_index_reader.py | 310 ++++++++++++++++++
.../globalindex/sorted_file_meta_selector.py | 169 ++++++++++
...ree_index_meta.py => sorted_index_file_meta.py} | 48 +--
.../globalindex/union_global_index_reader.py | 3 +
.../pypaimon/tests/btree_thread_safety_test.py | 76 ++++-
.../pypaimon/tests/e2e/java_py_read_write_test.py | 143 ++++++++
.../pypaimon/tests/sorted_index_file_meta_test.py | 72 ++++
.../pypaimon/tests/vector_search_filter_test.py | 12 +-
30 files changed, 1973 insertions(+), 801 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 1d2a4f3b85..2a7584a0bc 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -93,6 +93,7 @@ import static org.apache.paimon.CoreOptions.PATH;
import static org.apache.paimon.CoreOptions.ROW_TRACKING_ENABLED;
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.SimpleTableTestBase.getResult;
import static org.assertj.core.api.Assertions.assertThat;
@@ -515,6 +516,100 @@ public class JavaPyE2ETest {
testBtreeIndexWriteNull();
}
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testBitmapIndexWrite() throws Exception {
+ // create table
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.STRING(), DataTypes.STRING()},
+ new String[] {"k", "v"});
+ Options options = new Options();
+ Path tablePath = new Path(warehouse.toString() + "/default.db/test_bitmap_index_string");
+ LocalFileIO.create().delete(tablePath, true);
+ options.set(PATH, tablePath.toString());
+ options.set(ROW_TRACKING_ENABLED, true);
+ options.set(DATA_EVOLUTION_ENABLED, true);
+ options.set(GLOBAL_INDEX_ENABLED, true);
+ TableSchema tableSchema =
+ SchemaUtils.forceCommit(
+ new SchemaManager(LocalFileIO.create(), tablePath),
+ new Schema(
+ rowType.getFields(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options.toMap(),
+ ""));
+ AppendOnlyFileStoreTable table =
+ new AppendOnlyFileStoreTable(
+ FileIOFinder.find(tablePath),
+ tablePath,
+ tableSchema,
+ CatalogEnvironment.empty());
+
+ // write data
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(
+ GenericRow.of(BinaryString.fromString("k1"), BinaryString.fromString("v1")));
+ write.write(
+ GenericRow.of(BinaryString.fromString("k2"), BinaryString.fromString("v2")));
+ write.write(
+ GenericRow.of(BinaryString.fromString("k2"), BinaryString.fromString("v2b")));
+ write.write(
+ GenericRow.of(BinaryString.fromString("k3"), BinaryString.fromString("v3")));
+ write.write(GenericRow.of(null, BinaryString.fromString("v_null")));
+ commit.commit(write.prepareCommit());
+ }
+
+ // build index
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(table, "bitmap").withIndexField("k");
+ try (BatchTableCommit commit = writeBuilder.newCommit()) {
+ commit.commit(
+ builder.build(
+ builder.scan()
+ .map(org.apache.paimon.utils.Pair::getValue)
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Expected scan result when building index."))
+ .get(0),
+ IOManager.create(warehouse.toString())));
+ }
+
+ // assert index
+ List<IndexManifestEntry> indexEntries =
+ table.indexManifestFileReader().read(table.latestSnapshot().get().indexManifest);
+ assertThat(indexEntries)
+ .singleElement()
+ .matches(entry -> entry.indexFile().rowCount() == 5);
+
+ // read index
+ PredicateBuilder predicateBuilder = new PredicateBuilder(table.rowType());
+ ReadBuilder readBuilder =
+ table.newReadBuilder()
+ .withFilter(predicateBuilder.equal(0, BinaryString.fromString("k2")));
+ List<String> result = new ArrayList<>();
+ readBuilder
+ .newRead()
+ .createReader(readBuilder.newScan().plan())
+ .forEachRemaining(r -> result.add(r.getString(1).toString()));
+ assertThat(result).containsExactlyInAnyOrder("v2", "v2b");
+ }
+
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testCompressedGlobalIndexWrite() throws Exception {
+ for (String compression : Arrays.asList("lz4", "lzo")) {
+ writeCompressedGlobalIndexTable(
+ "btree", "test_btree_index_" + compression + "_fallback", compression);
+ writeCompressedGlobalIndexTable(
+ "bitmap", "test_bitmap_index_" + compression + "_fallback", compression);
+ }
+ }
+
private void testBtreeIndexWriteString() throws Exception {
testBtreeIndexWriteGeneric(
DataTypes.STRING(),
@@ -606,6 +701,90 @@ public class JavaPyE2ETest {
assertThat(result).containsOnly("v2");
}
+ private void writeCompressedGlobalIndexTable(
+ String indexType, String tableName, String compression) throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.STRING(), DataTypes.STRING()},
+ new String[] {"k", "v"});
+ Options options = new Options();
+ Path tablePath = new Path(warehouse.toString() + "/default.db/" + tableName);
+ LocalFileIO.create().delete(tablePath, true);
+ options.set(PATH, tablePath.toString());
+ options.set(ROW_TRACKING_ENABLED, true);
+ options.set(DATA_EVOLUTION_ENABLED, true);
+ options.set(GLOBAL_INDEX_ENABLED, true);
+ if ("btree".equals(indexType)) {
+ options.set(BTREE_INDEX_COMPRESSION, compression);
+ } else {
+ options.set(BITMAP_INDEX_COMPRESSION, compression);
+ }
+
+ TableSchema tableSchema =
+ SchemaUtils.forceCommit(
+ new SchemaManager(LocalFileIO.create(), tablePath),
+ new Schema(
+ rowType.getFields(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options.toMap(),
+ ""));
+ AppendOnlyFileStoreTable table =
+ new AppendOnlyFileStoreTable(
+ FileIOFinder.find(tablePath),
+ tablePath,
+ tableSchema,
+ CatalogEnvironment.empty());
+
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ for (int i = 0; i < 300; i++) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString(String.format("key-%03d", i)),
+ BinaryString.fromString("value-" + i)));
+ }
+ commit.commit(write.prepareCommit());
+ }
+
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(table, indexType).withIndexField("k");
+ try (BatchTableCommit commit = writeBuilder.newCommit()) {
+ commit.commit(
+ builder.build(
+ builder.scan()
+ .map(org.apache.paimon.utils.Pair::getValue)
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Expected scan result when building index."))
+ .get(0),
+ IOManager.create(warehouse.toString())));
+ }
+
+ List<IndexManifestEntry> indexEntries =
+ table.indexManifestFileReader().read(table.latestSnapshot().get().indexManifest);
+ assertThat(indexEntries)
+ .singleElement()
+ .matches(entry -> entry.indexFile().rowCount() == 300);
+
+ PredicateBuilder predicateBuilder = new PredicateBuilder(table.rowType());
+ ReadBuilder readBuilder =
+ table.newReadBuilder()
+ .withFilter(
+ predicateBuilder.greaterOrEqual(
+ 0, BinaryString.fromString("key-295")));
+ List<String> result = new ArrayList<>();
+ readBuilder
+ .newRead()
+ .createReader(readBuilder.newScan().plan())
+ .forEachRemaining(r -> result.add(r.getString(1).toString()));
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ "value-295", "value-296", "value-297", "value-298", "value-299");
+ }
+
private void testBtreeIndexWriteLarge() throws Exception {
// create table
RowType rowType =
diff --git a/paimon-python/dev/run_mixed_tests.sh b/paimon-python/dev/run_mixed_tests.sh
index 2014caa781..b2c8aec7a4 100755
--- a/paimon-python/dev/run_mixed_tests.sh
+++ b/paimon-python/dev/run_mixed_tests.sh
@@ -219,6 +219,53 @@ run_btree_index_test() {
fi
}
+run_bitmap_index_test() {
+ echo -e "${YELLOW}=== Step 6b: Running Bitmap Index Test (Java Write, Python Read) ===${NC}"
+
+ cd "$PROJECT_ROOT"
+
+ echo "Running Maven test for JavaPyE2ETest.testBitmapIndexWrite..."
+ if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testBitmapIndexWrite -pl paimon-core -q -Drun.e2e.tests=true; then
+ echo -e "${GREEN}✓ Java test completed successfully${NC}"
+ else
+ echo -e "${RED}✗ Java test failed${NC}"
+ return 1
+ fi
+ cd "$PAIMON_PYTHON_DIR"
+ # Run the specific Python test method
+ echo "Running Python test for JavaPyReadWriteTest.test_read_bitmap_index_table..."
+ if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest::test_read_bitmap_index_table -v; then
+ echo -e "${GREEN}✓ Python test completed successfully${NC}"
+ return 0
+ else
+ echo -e "${RED}✗ Python test failed${NC}"
+ return 1
+ fi
+}
+
+run_compressed_global_index_test() {
+ echo -e "${YELLOW}=== Step 6c: Running Compressed Global Index Test (Java Write, Python Read) ===${NC}"
+
+ cd "$PROJECT_ROOT"
+
+ echo "Running Maven test for JavaPyE2ETest.testCompressedGlobalIndexWrite..."
+ if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testCompressedGlobalIndexWrite -pl paimon-core -q -Drun.e2e.tests=true; then
+ echo -e "${GREEN}✓ Java test completed successfully${NC}"
+ else
+ echo -e "${RED}✗ Java test failed${NC}"
+ return 1
+ fi
+ cd "$PAIMON_PYTHON_DIR"
+ echo "Running Python test for JavaPyReadWriteTest.test_read_compressed_global_index_fallback_scan..."
+ if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest::test_read_compressed_global_index_fallback_scan -v; then
+ echo -e "${GREEN}✓ Python test completed successfully${NC}"
+ return 0
+ else
+ echo -e "${RED}✗ Python test failed${NC}"
+ return 1
+ fi
+}
+
run_compressed_text_test() {
echo -e "${YELLOW}=== Step 7: Running Compressed Text Test (Java Write, Python Read) ===${NC}"
@@ -691,6 +738,8 @@ main() {
local java_read_result=0
local pk_dv_result=0
local btree_index_result=0
+ local bitmap_index_result=0
+ local compressed_global_index_result=0
local compressed_text_result=0
local vector_append_table_result=0
local tantivy_fulltext_result=0
@@ -765,6 +814,19 @@ main() {
echo ""
+ # Run Bitmap index test (Java write, Python read)
+ if ! run_bitmap_index_test; then
+ bitmap_index_result=1
+ fi
+
+ echo ""
+
+ if ! run_compressed_global_index_test; then
+ compressed_global_index_result=1
+ fi
+
+ echo ""
+
if ! run_compressed_text_test; then
compressed_text_result=1
fi
@@ -940,6 +1002,18 @@ main() {
echo -e "${RED}✗ BTree Index Test (Java Write, Python Read): FAILED${NC}"
fi
+ if [[ $bitmap_index_result -eq 0 ]]; then
+ echo -e "${GREEN}✓ Bitmap Index Test (Java Write, Python Read): PASSED${NC}"
+ else
+ echo -e "${RED}✗ Bitmap Index Test (Java Write, Python Read): FAILED${NC}"
+ fi
+
+ if [[ $compressed_global_index_result -eq 0 ]]; then
+ echo -e "${GREEN}✓ Compressed Global Index Test (Java Write, Python Read): PASSED${NC}"
+ else
+ echo -e "${RED}✗ Compressed Global Index Test (Java Write, Python Read): FAILED${NC}"
+ fi
+
if [[ $compressed_text_result -eq 0 ]]; then
echo -e "${GREEN}✓ Compressed Text Test (Java Write, Python Read): PASSED${NC}"
else
@@ -1047,7 +1121,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 && $compressed_text_result -eq 0 && $tantivy_fulltext_result -eq 0 && $lumina_vector_result -eq 0 && $lumina_vector_btree_result -eq 0 && $compact_conflict_result -eq 0 && $blob_compact_conflict_result -eq 0 && $blob_alter_compact_result -eq 0 && $data_evolution_result -eq 0 && $data_evolution_py_write_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 && $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 && $compact_conflict_result -eq 0 && $blob_compact_conflict_result -eq 0 && $blob_alter_compact_result -eq 0 && $ [...]
echo -e "${GREEN}🎉 All tests passed! Java-Python interoperability verified.${NC}"
return 0
else
diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py
index 78c8db22e5..93b7b92b49 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -638,6 +638,24 @@ class CoreOptions:
)
)
+ BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE: ConfigOption[MemorySize] = (
+ ConfigOptions.key("btree-index.fallback-scan-max-size")
+ .memory_type()
+ .default_value(MemorySize.of_mebi_bytes(256))
+ .with_description(
+ "The maximum total BTree global index file size to allow fallback index scans."
+ )
+ )
+
+ BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE: ConfigOption[MemorySize] = (
+ ConfigOptions.key("bitmap-index.fallback-scan-max-size")
+ .memory_type()
+ .default_value(MemorySize.of_mebi_bytes(256))
+ .with_description(
+ "The maximum total bitmap global index file size to allow fallback dictionary scans."
+ )
+ )
+
LOCAL_CACHE_ENABLED: ConfigOption[bool] = (
ConfigOptions.key("local-cache.enabled")
.boolean_type()
@@ -1094,6 +1112,16 @@ class CoreOptions:
def global_index_thread_num(self) -> Optional[int]:
return self.options.get(CoreOptions.GLOBAL_INDEX_THREAD_NUM)
+ def btree_index_fallback_scan_max_size(self) -> int:
+ return self.options.get(
+ CoreOptions.BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE
+ ).get_bytes()
+
+ def bitmap_index_fallback_scan_max_size(self) -> int:
+ return self.options.get(
+ CoreOptions.BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE
+ ).get_bytes()
+
def local_cache_enabled(self) -> bool:
return self.options.get(CoreOptions.LOCAL_CACHE_ENABLED)
diff --git a/paimon-python/pypaimon/globalindex/__init__.py b/paimon-python/pypaimon/globalindex/__init__.py
index 86c000f60f..8cff38b24d 100644
--- a/paimon-python/pypaimon/globalindex/__init__.py
+++ b/paimon-python/pypaimon/globalindex/__init__.py
@@ -27,7 +27,12 @@ from pypaimon.globalindex.vector_search_result import (
from pypaimon.globalindex.global_index_meta import GlobalIndexMeta, GlobalIndexIOMeta
from pypaimon.globalindex.global_index_evaluator import GlobalIndexEvaluator
from pypaimon.globalindex.global_index_scanner import GlobalIndexScanner
+from pypaimon.globalindex.key_serializer import KeySerializer
+from pypaimon.globalindex.memory_slice_input import MemorySliceInput
from pypaimon.globalindex.offset_global_index_reader import OffsetGlobalIndexReader
+from pypaimon.globalindex.sorted_file_global_index_reader import SortedFileGlobalIndexReader
+from pypaimon.globalindex.sorted_file_meta_selector import SortedFileMetaSelector
+from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
from pypaimon.utils.range import Range
__all__ = [
@@ -43,6 +48,11 @@ __all__ = [
'GlobalIndexIOMeta',
'GlobalIndexEvaluator',
'GlobalIndexScanner',
+ 'KeySerializer',
+ 'MemorySliceInput',
'OffsetGlobalIndexReader',
+ 'SortedFileGlobalIndexReader',
+ 'SortedFileMetaSelector',
+ 'SortedIndexFileMeta',
'Range',
]
diff --git a/paimon-python/pypaimon/globalindex/bitmap/__init__.py b/paimon-python/pypaimon/globalindex/bitmap/__init__.py
new file mode 100644
index 0000000000..4e37cce1eb
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/bitmap/__init__.py
@@ -0,0 +1,26 @@
+# 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.
+
+"""Bitmap global index support."""
+
+from pypaimon.globalindex.bitmap.bitmap_index_reader import BitmapIndexReader
+from pypaimon.globalindex.bitmap.lazy_filtered_bitmap_reader import LazyFilteredBitmapReader
+
+__all__ = [
+ 'BitmapIndexReader',
+ 'LazyFilteredBitmapReader',
+]
diff --git a/paimon-python/pypaimon/globalindex/bitmap/bitmap_index_reader.py b/paimon-python/pypaimon/globalindex/bitmap/bitmap_index_reader.py
new file mode 100644
index 0000000000..0f182014c0
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/bitmap/bitmap_index_reader.py
@@ -0,0 +1,361 @@
+# 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.
+
+"""Reader for one bitmap global index file."""
+
+from dataclasses import dataclass
+import struct
+import threading
+from typing import Callable, Dict, List, Optional
+
+from pypaimon.common.file_io import FileIO, pread, supports_pread
+from pypaimon.globalindex.block_compression import (
+ BLOCK_TRAILER_LENGTH,
+ decompress_block_with_trailer,
+)
+from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
+from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.globalindex.key_serializer import KeySerializer
+from pypaimon.globalindex.memory_slice_input import MemorySliceInput
+from pypaimon.utils.roaring_bitmap import RoaringBitmap64
+
+
+@dataclass(frozen=True)
+class _BlockInfo:
+ offset: int
+ length: int
+
+
+@dataclass(frozen=True)
+class _DictionaryBlockMeta(_BlockInfo):
+ first_key: bytes
+
+
+@dataclass(frozen=True)
+class _DictionaryEntry:
+ key: bytes
+ bitmap_block: _BlockInfo
+
+
+@dataclass(frozen=True)
+class _Footer:
+ null_rows_block: _BlockInfo
+ non_null_rows_block: _BlockInfo
+ index_block: _BlockInfo
+
+
+class BitmapIndexReader:
+ """Synchronous reader for one bitmap global index file."""
+
+ _FOOTER_LENGTH = 48
+ _MAGIC = 0x42474958
+ _VERSION = 1
+
+ def __init__(
+ self,
+ key_serializer: KeySerializer,
+ file_io: FileIO,
+ index_path: str,
+ io_meta: GlobalIndexIOMeta,
+ ):
+ self._key_serializer = key_serializer
+ self._comparator = key_serializer.create_comparator()
+ self._io_meta = io_meta
+ file_path = (io_meta.external_path
+ if io_meta.external_path
+ else index_path + "/" + io_meta.file_name)
+ self._input_stream = file_io.new_input_stream(file_path)
+ self._supports_pread = supports_pread(self._input_stream)
+ self._io_lock = threading.Lock()
+
+ self._footer = self._read_footer()
+ self._null_rows: Optional[RoaringBitmap64] = None
+ self._non_null_rows: Optional[RoaringBitmap64] = None
+ self._dictionary_blocks: Optional[List[_DictionaryBlockMeta]] = None
+ self._dictionary_block_cache: Dict[_DictionaryBlockMeta, List[_DictionaryEntry]] = {}
+ self._lazy_lock = threading.Lock()
+
+ def _read_from(self, offset: int, length: int) -> bytes:
+ if self._supports_pread:
+ return pread(self._input_stream, length, offset)
+ with self._io_lock:
+ self._input_stream.seek(offset)
+ return self._input_stream.read(length)
+
+ def _read_footer(self) -> _Footer:
+ if self._io_meta.file_size < self._FOOTER_LENGTH:
+ raise ValueError("Invalid bitmap global index file size.")
+ footer_data = self._read_from(
+ self._io_meta.file_size - self._FOOTER_LENGTH,
+ self._FOOTER_LENGTH)
+ (null_offset, null_length,
+ non_null_offset, non_null_length,
+ index_offset, index_length,
+ value_count, version, magic) = struct.unpack('>q i q i q i i i i', footer_data)
+ if magic != self._MAGIC:
+ raise ValueError("File is not a bitmap global index file.")
+ if version != self._VERSION:
+ raise ValueError(f"Unsupported bitmap global index file version: {version}")
+ if value_count < 0:
+ raise ValueError("Invalid bitmap value count.")
+ return _Footer(
+ _BlockInfo(null_offset, null_length),
+ _BlockInfo(non_null_offset, non_null_length),
+ _BlockInfo(index_offset, index_length))
+
+ def _read_bitmap(self, block: _BlockInfo) -> RoaringBitmap64:
+ return RoaringBitmap64.deserialize(self._read_from(block.offset, block.length))
+
+ def _read_compressible_block(self, block: _BlockInfo) -> bytes:
+ block_and_trailer = self._read_from(
+ block.offset, block.length + BLOCK_TRAILER_LENGTH)
+ return decompress_block_with_trailer(block_and_trailer, block.length)
+
+ def _read_index_block(self) -> List[_DictionaryBlockMeta]:
+ data_input = MemorySliceInput(
+ self._read_compressible_block(self._footer.index_block))
+ block_count = data_input.read_var_len_int()
+ if block_count < 0:
+ raise ValueError("Invalid bitmap dictionary block count.")
+
+ blocks = []
+ for _ in range(block_count):
+ key_length = data_input.read_var_len_int()
+ if key_length < 0:
+ raise ValueError("Invalid bitmap key length.")
+ first_key = data_input.read_slice(key_length)
+ offset = data_input.read_var_len_long()
+ length = data_input.read_var_len_int()
+ blocks.append(_DictionaryBlockMeta(offset, length, first_key))
+ return sorted(blocks, key=lambda block: block.first_key)
+
+ def _read_dictionary_block(self, block: _DictionaryBlockMeta) -> List[_DictionaryEntry]:
+ data_input = MemorySliceInput(self._read_compressible_block(block))
+ entry_count = data_input.read_var_len_int()
+ if entry_count < 0:
+ raise ValueError("Invalid bitmap dictionary entry count.")
+
+ entries = []
+ for _ in range(entry_count):
+ key_length = data_input.read_var_len_int()
+ if key_length < 0:
+ raise ValueError("Invalid bitmap key length.")
+ key = data_input.read_slice(key_length)
+ bitmap_offset = data_input.read_var_len_long()
+ bitmap_length = data_input.read_var_len_int()
+ entries.append(_DictionaryEntry(key, _BlockInfo(bitmap_offset, bitmap_length)))
+ return entries
+
+ def _get_null_rows(self) -> RoaringBitmap64:
+ if self._null_rows is None:
+ with self._lazy_lock:
+ if self._null_rows is None:
+ self._null_rows = self._read_bitmap(self._footer.null_rows_block)
+ return self._copy(self._null_rows)
+
+ def _get_non_null_rows(self) -> RoaringBitmap64:
+ if self._non_null_rows is None:
+ with self._lazy_lock:
+ if self._non_null_rows is None:
+ self._non_null_rows = self._read_bitmap(self._footer.non_null_rows_block)
+ return self._copy(self._non_null_rows)
+
+ def _get_dictionary_blocks(self) -> List[_DictionaryBlockMeta]:
+ if self._dictionary_blocks is None:
+ with self._lazy_lock:
+ if self._dictionary_blocks is None:
+ self._dictionary_blocks = self._read_index_block()
+ return self._dictionary_blocks
+
+ def _get_dictionary_block(self, meta: _DictionaryBlockMeta) -> List[_DictionaryEntry]:
+ block = self._dictionary_block_cache.get(meta)
+ if block is None:
+ with self._lazy_lock:
+ block = self._dictionary_block_cache.get(meta)
+ if block is None:
+ block = self._read_dictionary_block(meta)
+ self._dictionary_block_cache[meta] = block
+ return block
+
+ def _find_dictionary_block_index(self, key: bytes) -> int:
+ blocks = self._get_dictionary_blocks()
+ low = 0
+ high = len(blocks) - 1
+ while low <= high:
+ mid = (low + high) >> 1
+ if blocks[mid].first_key <= key:
+ low = mid + 1
+ else:
+ high = mid - 1
+ return high
+
+ def _find_bitmap_block(self, key: bytes) -> Optional[_BlockInfo]:
+ blocks = self._get_dictionary_blocks()
+ if not blocks:
+ return None
+
+ index = self._find_dictionary_block_index(key)
+ if index < 0:
+ return None
+
+ for entry in self._get_dictionary_block(blocks[index]):
+ if entry.key == key:
+ return entry.bitmap_block
+ if entry.key > key:
+ return None
+ return None
+
+ def _scan_dictionary(self, predicate: Callable[[object], bool]) -> RoaringBitmap64:
+ result = RoaringBitmap64()
+ for block_meta in self._get_dictionary_blocks():
+ for entry in self._get_dictionary_block(block_meta):
+ key = self._key_serializer.deserialize(entry.key)
+ if predicate(key):
+ result = RoaringBitmap64.or_(result, self._read_bitmap(entry.bitmap_block))
+ return result
+
+ def _scan_serialized_dictionary(self, predicate: Callable[[bytes], bool]) -> RoaringBitmap64:
+ result = RoaringBitmap64()
+ for block_meta in self._get_dictionary_blocks():
+ for entry in self._get_dictionary_block(block_meta):
+ if predicate(entry.key):
+ result = RoaringBitmap64.or_(result, self._read_bitmap(entry.bitmap_block))
+ return result
+
+ def visit_is_not_null(self) -> Optional[GlobalIndexResult]:
+ return GlobalIndexResult.create(self._get_non_null_rows())
+
+ def visit_is_null(self) -> Optional[GlobalIndexResult]:
+ return GlobalIndexResult.create(self._get_null_rows())
+
+ def visit_starts_with(self, literal: object) -> Optional[GlobalIndexResult]:
+ prefix = self._key_serializer.serialize(literal)
+ if len(prefix) == 0:
+ return GlobalIndexResult.create(self._get_non_null_rows())
+ return GlobalIndexResult.create(self._scan_serialized_dictionary(
+ lambda key: key.startswith(prefix)))
+
+ def visit_ends_with(self, literal: object) -> Optional[GlobalIndexResult]:
+ suffix = self._key_serializer.serialize(literal)
+ if len(suffix) == 0:
+ return GlobalIndexResult.create(self._get_non_null_rows())
+ return GlobalIndexResult.create(self._scan_serialized_dictionary(
+ lambda key: key.endswith(suffix)))
+
+ def visit_contains(self, literal: object) -> Optional[GlobalIndexResult]:
+ infix = self._key_serializer.serialize(literal)
+ if len(infix) == 0:
+ return GlobalIndexResult.create(self._get_non_null_rows())
+ return GlobalIndexResult.create(self._scan_serialized_dictionary(
+ lambda key: infix in key))
+
+ def visit_like(self, literal: object) -> Optional[GlobalIndexResult]:
+ from pypaimon.common.predicate import Like
+ return GlobalIndexResult.create(
+ self.like(lambda key: Like().test_by_value(key, [literal])))
+
+ def visit_less_than(self, literal: object) -> Optional[GlobalIndexResult]:
+ if literal is None:
+ return GlobalIndexResult.create_empty()
+ return GlobalIndexResult.create(
+ self._scan_dictionary(lambda key: self._comparator(key, literal) < 0))
+
+ def visit_greater_or_equal(self, literal: object) -> Optional[GlobalIndexResult]:
+ if literal is None:
+ return GlobalIndexResult.create_empty()
+ return GlobalIndexResult.create(
+ self._scan_dictionary(lambda key: self._comparator(key, literal) >= 0))
+
+ def visit_not_equal(self, literal: object) -> Optional[GlobalIndexResult]:
+ if literal is None:
+ return GlobalIndexResult.create_empty()
+ return GlobalIndexResult.create(
+ RoaringBitmap64.remove_all(self._get_non_null_rows(), self._equal(literal)))
+
+ def visit_less_or_equal(self, literal: object) -> Optional[GlobalIndexResult]:
+ if literal is None:
+ return GlobalIndexResult.create_empty()
+ return GlobalIndexResult.create(
+ self._scan_dictionary(lambda key: self._comparator(key, literal) <= 0))
+
+ def visit_equal(self, literal: object) -> Optional[GlobalIndexResult]:
+ return GlobalIndexResult.create(self._equal(literal))
+
+ def visit_greater_than(self, literal: object) -> Optional[GlobalIndexResult]:
+ if literal is None:
+ return GlobalIndexResult.create_empty()
+ return GlobalIndexResult.create(
+ self._scan_dictionary(lambda key: self._comparator(key, literal) > 0))
+
+ def visit_in(self, literals: List[object]) -> Optional[GlobalIndexResult]:
+ result = RoaringBitmap64()
+ seen = set()
+ for literal in literals:
+ if literal is None:
+ continue
+ key = self._key_serializer.serialize(literal)
+ if key in seen:
+ continue
+ seen.add(key)
+ bitmap_block = self._find_bitmap_block(key)
+ if bitmap_block is not None:
+ result = RoaringBitmap64.or_(result, self._read_bitmap(bitmap_block))
+ return GlobalIndexResult.create(result)
+
+ def visit_not_in(self, literals: List[object]) -> Optional[GlobalIndexResult]:
+ if any(literal is None for literal in literals):
+ return GlobalIndexResult.create_empty()
+ result = self._get_non_null_rows()
+ in_result = self.visit_in(literals).results()
+ return GlobalIndexResult.create(RoaringBitmap64.remove_all(result, in_result))
+
+ def visit_between(self, from_v: object, to_v: object) -> Optional[GlobalIndexResult]:
+ if from_v is None or to_v is None or self._comparator(from_v, to_v) > 0:
+ return GlobalIndexResult.create_empty()
+ return GlobalIndexResult.create(
+ self._scan_dictionary(
+ lambda key: self._comparator(key, from_v) >= 0
+ and self._comparator(key, to_v) <= 0))
+
+ def like(self, key_predicate: Callable[[object], bool]) -> RoaringBitmap64:
+ return self._scan_dictionary(key_predicate)
+
+ def less_than(self, literal: object) -> RoaringBitmap64:
+ if literal is None:
+ return RoaringBitmap64()
+ return self._scan_dictionary(lambda key: self._comparator(key, literal) < 0)
+
+ def greater_than(self, literal: object) -> RoaringBitmap64:
+ if literal is None:
+ return RoaringBitmap64()
+ return self._scan_dictionary(lambda key: self._comparator(key, literal) > 0)
+
+ def _equal(self, literal: object) -> RoaringBitmap64:
+ if literal is None:
+ return RoaringBitmap64()
+ bitmap_block = self._find_bitmap_block(self._key_serializer.serialize(literal))
+ if bitmap_block is None:
+ return RoaringBitmap64()
+ return self._read_bitmap(bitmap_block)
+
+ @staticmethod
+ def _copy(bitmap: RoaringBitmap64) -> RoaringBitmap64:
+ return RoaringBitmap64.or_(RoaringBitmap64(), bitmap)
+
+ def close(self) -> None:
+ if self._input_stream is not None:
+ self._input_stream.close()
diff --git a/paimon-python/pypaimon/globalindex/bitmap/lazy_filtered_bitmap_reader.py b/paimon-python/pypaimon/globalindex/bitmap/lazy_filtered_bitmap_reader.py
new file mode 100644
index 0000000000..024f719cbd
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/bitmap/lazy_filtered_bitmap_reader.py
@@ -0,0 +1,34 @@
+# 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.
+
+"""Bitmap global index reader for sorted index files."""
+
+from pypaimon.globalindex.bitmap.bitmap_index_reader import BitmapIndexReader
+from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
+from pypaimon.globalindex.sorted_file_global_index_reader import SortedFileGlobalIndexReader
+
+
+class LazyFilteredBitmapReader(SortedFileGlobalIndexReader):
+ """Manages multiple bitmap index files for one row-id range."""
+
+ def open_reader(self, meta: GlobalIndexIOMeta) -> BitmapIndexReader:
+ return BitmapIndexReader(
+ key_serializer=self._key_serializer,
+ file_io=self._file_io,
+ index_path=self._index_path,
+ io_meta=meta,
+ )
diff --git a/paimon-python/pypaimon/globalindex/block_compression.py b/paimon-python/pypaimon/globalindex/block_compression.py
new file mode 100644
index 0000000000..68931af6e3
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/block_compression.py
@@ -0,0 +1,319 @@
+# 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.
+
+"""Shared block decompression for Java global-index block trailers."""
+
+import struct
+import zlib
+
+from pypaimon.globalindex.memory_slice_input import MemorySliceInput
+
+
+COMPRESSION_NONE = 0
+COMPRESSION_ZSTD = 1
+COMPRESSION_LZ4 = 2
+COMPRESSION_LZO = 3
+BLOCK_TRAILER_LENGTH = 5
+_HEADER_LENGTH = 8
+
+
+def crc32c(data: bytes, compression_type: int) -> int:
+ crc = zlib.crc32(data)
+ crc = zlib.crc32(bytes([compression_type & 0xFF]), crc)
+ return crc & 0xFFFFFFFF
+
+
+def decompress_block(block_bytes: bytes, compression_type: int) -> bytes:
+ if compression_type == COMPRESSION_NONE:
+ return block_bytes
+ if compression_type == COMPRESSION_ZSTD:
+ return _decompress_zstd(block_bytes)
+ if compression_type == COMPRESSION_LZ4:
+ return _decompress_lz4(block_bytes)
+ if compression_type == COMPRESSION_LZO:
+ return _decompress_lzo(block_bytes)
+ raise ValueError("Compression type %s not supported" % compression_type)
+
+
+def decompress_block_with_trailer(block_and_trailer: bytes, block_length: int) -> bytes:
+ if len(block_and_trailer) < block_length + BLOCK_TRAILER_LENGTH:
+ raise ValueError("Block data too short to contain trailer.")
+
+ block_bytes = block_and_trailer[:block_length]
+ trailer = block_and_trailer[block_length:block_length + BLOCK_TRAILER_LENGTH]
+ compression_type = trailer[0]
+ expected_crc = struct.unpack('<I', trailer[1:5])[0]
+ actual_crc = crc32c(block_bytes, compression_type)
+ if actual_crc != expected_crc:
+ raise ValueError(
+ "CRC32 mismatch: expected %s, got %s" % (expected_crc, actual_crc))
+ return decompress_block(block_bytes, compression_type)
+
+
+def _decompress_zstd(block_bytes: bytes) -> bytes:
+ import zstandard as zstd
+ from io import BytesIO
+
+ memory_input = MemorySliceInput(block_bytes)
+ expected_len = memory_input.read_var_len_int()
+ compressed = block_bytes[memory_input.position():]
+ with zstd.ZstdDecompressor().stream_reader(BytesIO(compressed)) as reader:
+ uncompressed = reader.read()
+ if len(uncompressed) != expected_len:
+ raise ValueError("Corrupted block, decompression failed.")
+ return uncompressed
+
+
+def _decompress_lz4(block_bytes: bytes) -> bytes:
+ memory_input = MemorySliceInput(block_bytes)
+ expected_len = memory_input.read_var_len_int()
+ payload_start = memory_input.position()
+ if len(block_bytes) - payload_start < _HEADER_LENGTH:
+ raise ValueError("Compressed block data too short to contain header.")
+ compressed_len, original_len = struct.unpack_from('<ii', block_bytes, payload_start)
+ _validate_lengths(compressed_len, original_len)
+ if original_len != expected_len:
+ raise ValueError("Compressed block header has unexpected original length.")
+ payload_start += _HEADER_LENGTH
+ payload_end = payload_start + compressed_len
+ if len(block_bytes) < payload_end:
+ raise ValueError("Compressed block data is incomplete.")
+
+ payload = block_bytes[payload_start:payload_end]
+ uncompressed = _decompress_lz4_raw(payload, original_len)
+ if len(uncompressed) != original_len:
+ raise ValueError("Corrupted block, decompression failed.")
+ return uncompressed
+
+
+def _decompress_lzo(block_bytes: bytes) -> bytes:
+ memory_input = MemorySliceInput(block_bytes)
+ expected_len = memory_input.read_var_len_int()
+ payload_start = memory_input.position()
+ if len(block_bytes) - payload_start < _HEADER_LENGTH:
+ raise ValueError("Compressed block data too short to contain header.")
+ compressed_len, original_len = struct.unpack_from('<ii', block_bytes, payload_start)
+ _validate_lengths(compressed_len, original_len)
+ if original_len != expected_len:
+ raise ValueError("Compressed block header has unexpected original length.")
+ payload_start += _HEADER_LENGTH
+ payload_end = payload_start + compressed_len
+ if len(block_bytes) < payload_end:
+ raise ValueError("Compressed block data is incomplete.")
+
+ payload = block_bytes[payload_start:payload_end]
+ uncompressed = _decompress_lzo_raw(payload, original_len)
+ if len(uncompressed) != original_len:
+ raise ValueError("Corrupted block, decompression failed.")
+ return uncompressed
+
+
+def _decompress_lz4_raw(payload: bytes, output_len: int) -> bytes:
+ output = bytearray()
+ pos = 0
+ payload_len = len(payload)
+
+ while pos < payload_len:
+ token = payload[pos]
+ pos += 1
+
+ literal_len = token >> 4
+ if literal_len == 15:
+ length_part, pos = _read_lz4_length(payload, pos, payload_len)
+ literal_len += length_part
+ literal_end = pos + literal_len
+ if literal_end > payload_len:
+ raise ValueError("Corrupted LZ4 block, literal exceeds input.")
+ output.extend(payload[pos:literal_end])
+ pos = literal_end
+
+ if pos == payload_len:
+ break
+ if pos + 2 > payload_len:
+ raise ValueError("Corrupted LZ4 block, missing match offset.")
+ offset = payload[pos] | (payload[pos + 1] << 8)
+ pos += 2
+ if offset <= 0 or offset > len(output):
+ raise ValueError("Corrupted LZ4 block, invalid match offset.")
+
+ match_len = token & 0x0F
+ if match_len == 15:
+ length_part, pos = _read_lz4_length(payload, pos, payload_len)
+ match_len += length_part
+ match_len += 4
+
+ start = len(output) - offset
+ for _ in range(match_len):
+ output.append(output[start])
+ start += 1
+ if len(output) > output_len:
+ raise ValueError("Corrupted LZ4 block, output exceeds expected length.")
+
+ if len(output) != output_len:
+ raise ValueError("Corrupted LZ4 block, unexpected output length.")
+ return bytes(output)
+
+
+def _decompress_lzo_raw(payload: bytes, output_len: int) -> bytes:
+ if not payload:
+ if output_len == 0:
+ return b''
+ raise ValueError("Corrupted LZO block, empty payload.")
+
+ output = bytearray()
+ pos = 0
+ payload_len = len(payload)
+ first_command = True
+ last_literal_len = 0
+
+ while pos < payload_len:
+ first_command = True
+ last_literal_len = 0
+ while True:
+ if pos >= payload_len:
+ raise ValueError("Corrupted LZO block, missing command.")
+ command = payload[pos]
+ pos += 1
+
+ match_len = 0
+ match_offset = 0
+
+ if (command & 0xF0) == 0:
+ if last_literal_len == 0:
+ literal_len = command & 0x0F
+ if literal_len == 0:
+ literal_len, pos = _read_lzo_variable_length(
+ payload, pos, payload_len, 0x0F)
+ literal_len += 3
+ elif last_literal_len <= 3:
+ if pos >= payload_len:
+ raise ValueError("Corrupted LZO block, missing match offset.")
+ match_len = 2
+ match_offset = ((command & 0x0C) >> 2) | (payload[pos] << 2)
+ pos += 1
+ literal_len = command & 0x03
+ else:
+ if pos >= payload_len:
+ raise ValueError("Corrupted LZO block, missing match offset.")
+ match_len = 3
+ match_offset = (((command & 0x0C) >> 2)
+ | (payload[pos] << 2)
+ | 0x0800)
+ pos += 1
+ literal_len = command & 0x03
+ elif first_command:
+ literal_len = command - 17
+ elif (command & 0xF0) == 0x10:
+ match_len = command & 0x07
+ if match_len == 0:
+ match_len, pos = _read_lzo_variable_length(
+ payload, pos, payload_len, 0x07)
+ match_len += 2
+ trailer, pos = _read_lzo_short(payload, pos, payload_len)
+ match_offset = ((command & 0x08) << 11) + (trailer >> 2)
+ if match_offset == 0:
+ break
+ match_offset += 0x3FFF
+ literal_len = trailer & 0x03
+ elif (command & 0xE0) == 0x20:
+ match_len = command & 0x1F
+ if match_len == 0:
+ match_len, pos = _read_lzo_variable_length(
+ payload, pos, payload_len, 0x1F)
+ match_len += 2
+ trailer, pos = _read_lzo_short(payload, pos, payload_len)
+ match_offset = trailer >> 2
+ literal_len = trailer & 0x03
+ elif (command & 0xC0) != 0:
+ if pos >= payload_len:
+ raise ValueError("Corrupted LZO block, missing match offset.")
+ match_len = ((command & 0xE0) >> 5) + 1
+ match_offset = ((command & 0x1C) >> 2) | (payload[pos] << 3)
+ pos += 1
+ literal_len = command & 0x03
+ else:
+ raise ValueError("Corrupted LZO block, invalid command.")
+
+ first_command = False
+ if match_len < 0 or literal_len < 0:
+ raise ValueError("Corrupted LZO block, invalid length.")
+
+ if match_len:
+ _copy_lzo_match(output, match_offset + 1, match_len, output_len)
+
+ literal_end = pos + literal_len
+ if literal_end > payload_len:
+ raise ValueError("Corrupted LZO block, literal exceeds input.")
+ output.extend(payload[pos:literal_end])
+ pos = literal_end
+ if len(output) > output_len:
+ raise ValueError("Corrupted LZO block, output exceeds expected length.")
+ last_literal_len = literal_len
+
+ if len(output) != output_len:
+ raise ValueError("Corrupted LZO block, unexpected output length.")
+ return bytes(output)
+
+
+def _read_lzo_variable_length(payload: bytes, pos: int, payload_len: int, base: int):
+ length = base
+ while True:
+ if pos >= payload_len:
+ raise ValueError("Corrupted LZO block, malformed length.")
+ value = payload[pos]
+ pos += 1
+ if value != 0:
+ return length + value, pos
+ length += 255
+
+
+def _read_lzo_short(payload: bytes, pos: int, payload_len: int):
+ if pos + 2 > payload_len:
+ raise ValueError("Corrupted LZO block, missing match trailer.")
+ return payload[pos] | (payload[pos + 1] << 8), pos + 2
+
+
+def _copy_lzo_match(output: bytearray, offset: int, match_len: int, output_len: int) -> None:
+ if offset <= 0 or offset > len(output):
+ raise ValueError("Corrupted LZO block, invalid match offset.")
+ if len(output) + match_len > output_len:
+ raise ValueError("Corrupted LZO block, output exceeds expected length.")
+
+ start = len(output) - offset
+ for _ in range(match_len):
+ output.append(output[start])
+ start += 1
+
+
+def _read_lz4_length(payload: bytes, pos: int, payload_len: int):
+ length = 0
+ while True:
+ if pos >= payload_len:
+ raise ValueError("Corrupted LZ4 block, malformed length.")
+ value = payload[pos]
+ pos += 1
+ length += value
+ if value != 255:
+ return length, pos
+
+
+def _validate_lengths(compressed_len: int, original_len: int) -> None:
+ if (original_len < 0
+ or compressed_len < 0
+ or (original_len == 0 and compressed_len != 0)
+ or (original_len != 0 and compressed_len == 0)):
+ raise ValueError("Input is corrupted, invalid length.")
diff --git a/paimon-python/pypaimon/globalindex/btree/block_reader.py b/paimon-python/pypaimon/globalindex/btree/block_reader.py
index b32012abf9..1702696943 100644
--- a/paimon-python/pypaimon/globalindex/btree/block_reader.py
+++ b/paimon-python/pypaimon/globalindex/btree/block_reader.py
@@ -22,7 +22,7 @@ from typing import Callable, Optional
from pypaimon.globalindex.btree.block_aligned_type import BlockAlignedType
from pypaimon.globalindex.btree.block_entry import BlockEntry
-from pypaimon.globalindex.btree.memory_slice_input import MemorySliceInput
+from pypaimon.globalindex.memory_slice_input import MemorySliceInput
class BlockReader:
diff --git a/paimon-python/pypaimon/globalindex/btree/btree_file_meta_selector.py b/paimon-python/pypaimon/globalindex/btree/btree_file_meta_selector.py
index 3770d842aa..7b00b59079 100644
--- a/paimon-python/pypaimon/globalindex/btree/btree_file_meta_selector.py
+++ b/paimon-python/pypaimon/globalindex/btree/btree_file_meta_selector.py
@@ -15,118 +15,9 @@
# specific language governing permissions and limitations
# under the License.
-"""Prunes BTree index files by comparing predicate literals against file min/max keys."""
+"""Compatibility alias for sorted global index file pruning."""
-from typing import List, Optional, Tuple
+from pypaimon.globalindex.sorted_file_meta_selector import SortedFileMetaSelector
-from pypaimon.globalindex.btree.btree_index_meta import BTreeIndexMeta
-from pypaimon.globalindex.btree.key_serializer import KeySerializer
-from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
-
-class BTreeFileMetaSelector:
- """Selects which BTree index files may contain rows matching a predicate.
-
- Each select_* method returns:
- - None: cannot evaluate (caller must visit all files)
- - []: all files pruned (result is empty)
- - [...]: only these files need to be visited
- """
-
- def __init__(self, files: List[Tuple[GlobalIndexIOMeta, BTreeIndexMeta]],
- key_serializer: KeySerializer):
- self._files = files
- self._key_serializer = key_serializer
- self._comparator = key_serializer.create_comparator()
-
- def _filter(self, predicate) -> List[GlobalIndexIOMeta]:
- return [io_meta for io_meta, idx_meta in self._files if predicate(idx_meta)]
-
- def _all_non_null_files(self) -> List[GlobalIndexIOMeta]:
- return self._filter(lambda m: not m.only_nulls())
-
- def select_is_null(self) -> List[GlobalIndexIOMeta]:
- return self._filter(lambda m: m.has_nulls)
-
- def select_is_not_null(self) -> List[GlobalIndexIOMeta]:
- return self._filter(lambda m: not m.only_nulls())
-
- def select_equal(self, literal) -> List[GlobalIndexIOMeta]:
- def pred(meta):
- if meta.only_nulls():
- return False
- first = self._key_serializer.deserialize(meta.first_key)
- last = self._key_serializer.deserialize(meta.last_key)
- return self._comparator(literal, first) >= 0 and self._comparator(literal, last) <= 0
- return self._filter(pred)
-
- def select_not_equal(self, literal) -> Optional[List[GlobalIndexIOMeta]]:
- return None
-
- def select_less_than(self, literal) -> List[GlobalIndexIOMeta]:
- def pred(meta):
- if meta.only_nulls():
- return False
- first = self._key_serializer.deserialize(meta.first_key)
- return self._comparator(first, literal) < 0
- return self._filter(pred)
-
- def select_less_or_equal(self, literal) -> List[GlobalIndexIOMeta]:
- def pred(meta):
- if meta.only_nulls():
- return False
- first = self._key_serializer.deserialize(meta.first_key)
- return self._comparator(first, literal) <= 0
- return self._filter(pred)
-
- def select_greater_than(self, literal) -> List[GlobalIndexIOMeta]:
- def pred(meta):
- if meta.only_nulls():
- return False
- last = self._key_serializer.deserialize(meta.last_key)
- return self._comparator(last, literal) > 0
- return self._filter(pred)
-
- def select_greater_or_equal(self, literal) -> List[GlobalIndexIOMeta]:
- def pred(meta):
- if meta.only_nulls():
- return False
- last = self._key_serializer.deserialize(meta.last_key)
- return self._comparator(last, literal) >= 0
- return self._filter(pred)
-
- def select_in(self, literals) -> List[GlobalIndexIOMeta]:
- def pred(meta):
- if meta.only_nulls():
- return False
- first = self._key_serializer.deserialize(meta.first_key)
- last = self._key_serializer.deserialize(meta.last_key)
- for lit in literals:
- if self._comparator(lit, first) >= 0 and self._comparator(lit, last) <= 0:
- return True
- return False
- return self._filter(pred)
-
- def select_not_in(self, literals) -> Optional[List[GlobalIndexIOMeta]]:
- return None
-
- def select_between(self, from_v, to_v) -> List[GlobalIndexIOMeta]:
- def pred(meta):
- if meta.only_nulls():
- return False
- first = self._key_serializer.deserialize(meta.first_key)
- last = self._key_serializer.deserialize(meta.last_key)
- return self._comparator(from_v, last) <= 0 and self._comparator(to_v, first) >= 0
- return self._filter(pred)
-
- def select_starts_with(self, literal) -> Optional[List[GlobalIndexIOMeta]]:
- return None
-
- def select_ends_with(self, literal) -> Optional[List[GlobalIndexIOMeta]]:
- return None
-
- def select_contains(self, literal) -> Optional[List[GlobalIndexIOMeta]]:
- return None
-
- def select_like(self, literal) -> Optional[List[GlobalIndexIOMeta]]:
- return None
+BTreeFileMetaSelector = SortedFileMetaSelector
diff --git a/paimon-python/pypaimon/globalindex/btree/btree_index_meta.py b/paimon-python/pypaimon/globalindex/btree/btree_index_meta.py
index ce12146c20..c482266e52 100644
--- a/paimon-python/pypaimon/globalindex/btree/btree_index_meta.py
+++ b/paimon-python/pypaimon/globalindex/btree/btree_index_meta.py
@@ -15,48 +15,9 @@
# specific language governing permissions and limitations
# under the License.
-"""B-tree index metadata."""
+"""Compatibility alias for sorted global index file metadata."""
-from dataclasses import dataclass
-from typing import Optional
-import struct
+from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
-@dataclass
-class BTreeIndexMeta:
- """
- Index Meta of each BTree index file.
-
- The first key and last key of this meta could be null if the
- entire btree index file only contains nulls.
- """
-
- first_key: Optional[bytes]
- last_key: Optional[bytes]
- has_nulls: bool
-
- def only_nulls(self) -> bool:
- """Check if this index only contains nulls."""
- return self.first_key is None and self.last_key is None
-
- @classmethod
- def deserialize(cls, data: bytes) -> 'BTreeIndexMeta':
- """Deserialize metadata from byte array."""
- offset = 0
-
- # Read first key
- first_key_length = struct.unpack('<I', data[offset:offset + 4])[0]
- offset += 4
- first_key = None if first_key_length == 0 else data[offset:offset + first_key_length]
- offset += first_key_length
-
- # Read last key
- last_key_length = struct.unpack('<I', data[offset:offset + 4])[0]
- offset += 4
- last_key = None if last_key_length == 0 else data[offset:offset + last_key_length]
- offset += last_key_length
-
- # Read has_nulls flag
- has_nulls = struct.unpack('<B', data[offset:offset + 1])[0] == 1
-
- return cls(first_key, last_key, has_nulls)
+BTreeIndexMeta = SortedIndexFileMeta
diff --git a/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py b/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py
index 5a27a24ecc..61decbefef 100644
--- a/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py
+++ b/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py
@@ -28,13 +28,13 @@ from typing import List, Optional
from pypaimon.common.file_io import FileIO, supports_pread, pread
from pypaimon.globalindex.btree.btree_index_meta import BTreeIndexMeta
-from pypaimon.globalindex.btree.key_serializer import KeySerializer
from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.globalindex.key_serializer import KeySerializer
from pypaimon.utils.roaring_bitmap import RoaringBitmap64
from pypaimon.globalindex.btree.btree_file_footer import BTreeFileFooter
from pypaimon.globalindex.btree.sst_file_reader import SstFileReader
-from pypaimon.globalindex.btree.memory_slice_input import MemorySliceInput
+from pypaimon.globalindex.memory_slice_input import MemorySliceInput
def _deserialize_row_ids(data: bytes) -> List[int]:
@@ -200,8 +200,7 @@ class BTreeIndexReader:
return GlobalIndexResult.create(self._read_null_bitmap())
def visit_less_than(self, literal: object) -> Optional[GlobalIndexResult]:
- return GlobalIndexResult.create(
- self._range_query(self.min_key, literal, True, False))
+ return GlobalIndexResult.create(self.less_than(literal))
def visit_greater_or_equal(self, literal: object) -> Optional[GlobalIndexResult]:
return GlobalIndexResult.create(
@@ -222,8 +221,7 @@ class BTreeIndexReader:
self._range_query(literal, literal, True, True))
def visit_greater_than(self, literal: object) -> Optional[GlobalIndexResult]:
- return GlobalIndexResult.create(
- self._range_query(literal, self.max_key, False, True))
+ return GlobalIndexResult.create(self.greater_than(literal))
def visit_in(self, literals: List[object]) -> Optional[GlobalIndexResult]:
result = RoaringBitmap64()
@@ -255,6 +253,12 @@ class BTreeIndexReader:
return GlobalIndexResult.create(
self._range_query(min_v, max_v, True, True))
+ def less_than(self, literal: object) -> RoaringBitmap64:
+ return self._range_query(self.min_key, literal, True, False)
+
+ def greater_than(self, literal: object) -> RoaringBitmap64:
+ return self._range_query(literal, self.max_key, False, True)
+
def close(self) -> None:
if self.input_stream is not None:
self.input_stream.close()
diff --git a/paimon-python/pypaimon/globalindex/btree/key_serializer.py b/paimon-python/pypaimon/globalindex/btree/key_serializer.py
index 58f35bab25..f94a2ae955 100644
--- a/paimon-python/pypaimon/globalindex/btree/key_serializer.py
+++ b/paimon-python/pypaimon/globalindex/btree/key_serializer.py
@@ -15,130 +15,20 @@
# specific language governing permissions and limitations
# under the License.
-"""Key serializer for B-tree index."""
-
-from abc import ABC, abstractmethod
-from typing import Callable
-import struct
-
-from pypaimon.schema.data_types import DataType
-from pypaimon.schema.data_types import AtomicType
-
-
-class KeySerializer(ABC):
- """
- Interface for serializing and deserializing B-tree index keys.
-
- This interface provides core methods to ser/de and compare btree index keys.
- """
-
- @abstractmethod
- def serialize(self, key: object) -> bytes:
- """Serialize a key to bytes."""
- pass
-
- @abstractmethod
- def deserialize(self, data: bytes) -> object:
- """Deserialize bytes to a key."""
- pass
-
- @abstractmethod
- def create_comparator(self) -> Callable[[object, object], int]:
- """
- Create a comparator function for keys.
-
- Returns:
- A function that takes two keys and returns:
- - negative if first < second
- - 0 if first == second
- - positive if first > second
- """
- pass
-
-
-class StringSerializer(KeySerializer):
- """Serializer for STRING type."""
-
- def serialize(self, key: object) -> bytes:
- """Serialize a string key to bytes."""
- if isinstance(key, str):
- return key.encode('utf-8')
- return str(key).encode('utf-8')
-
- def deserialize(self, data: bytes) -> object:
- """Deserialize bytes to a string key."""
- return data.decode('utf-8')
-
- def create_comparator(self) -> Callable[[object, object], int]:
- """Create a comparator for string keys."""
- def compare(a: object, b: object) -> int:
- str_a = a if isinstance(a, str) else str(a)
- str_b = b if isinstance(b, str) else str(b)
- if str_a < str_b:
- return -1
- elif str_a > str_b:
- return 1
- return 0
- return compare
-
-
-class LongSerializer(KeySerializer):
- """Serializer for BIGINT type."""
-
- def serialize(self, key: object) -> bytes:
- """Serialize a long key to bytes."""
- return struct.pack('<q', int(key))
-
- def deserialize(self, data: bytes) -> object:
- """Deserialize bytes to a long key."""
- return struct.unpack('<q', data)[0]
-
- def create_comparator(self) -> Callable[[object, object], int]:
- """Create a comparator for long keys."""
- def compare(a: object, b: object) -> int:
- long_a = int(a)
- long_b = int(b)
- if long_a < long_b:
- return -1
- elif long_a > long_b:
- return 1
- return 0
- return compare
-
-
-class IntSerializer(KeySerializer):
- """Serializer for INT type."""
-
- def serialize(self, key: object) -> bytes:
- """Serialize an int key to bytes."""
- return struct.pack('<i', int(key))
-
- def deserialize(self, data: bytes) -> object:
- """Deserialize bytes to an int key."""
- return struct.unpack('<i', data)[0]
-
- def create_comparator(self) -> Callable[[object, object], int]:
- """Create a comparator for int keys."""
- def compare(a: object, b: object) -> int:
- int_a = int(a)
- int_b = int(b)
- if int_a < int_b:
- return -1
- elif int_a > int_b:
- return 1
- return 0
- return compare
-
-
-def create_serializer(data_type: DataType) -> KeySerializer:
- if not isinstance(data_type, AtomicType):
- raise ValueError(f"Key serializer only support AtomicType yet, meet {data_type.__class__}")
- type_name = data_type.type.upper()
- if type_name in ('CHAR', 'VARCHAR', 'STRING'):
- return StringSerializer()
- elif type_name == 'BIGINT':
- return LongSerializer()
- elif type_name == 'INT':
- return IntSerializer()
- else:
- raise ValueError(f"DataType: {data_type} is not supported by btree index now.")
+"""Compatibility re-export for the global index key serializer."""
+
+from pypaimon.globalindex.key_serializer import (
+ IntSerializer,
+ KeySerializer,
+ LongSerializer,
+ StringSerializer,
+ create_serializer,
+)
+
+__all__ = [
+ 'IntSerializer',
+ 'KeySerializer',
+ 'LongSerializer',
+ 'StringSerializer',
+ 'create_serializer',
+]
diff --git a/paimon-python/pypaimon/globalindex/btree/lazy_filtered_btree_reader.py b/paimon-python/pypaimon/globalindex/btree/lazy_filtered_btree_reader.py
index 8dc19e7151..23bc1e1d98 100644
--- a/paimon-python/pypaimon/globalindex/btree/lazy_filtered_btree_reader.py
+++ b/paimon-python/pypaimon/globalindex/btree/lazy_filtered_btree_reader.py
@@ -15,196 +15,20 @@
# specific language governing permissions and limitations
# under the License.
-"""A GlobalIndexReader that manages multiple BTree files with lazy reader creation and file pruning."""
+"""BTree global index reader for sorted index files."""
-import threading
-from concurrent.futures import Executor, Future
-from typing import Callable, Dict, List, Optional
-
-from pypaimon.globalindex.btree.btree_file_meta_selector import BTreeFileMetaSelector
-from pypaimon.globalindex.btree.btree_index_meta import BTreeIndexMeta
from pypaimon.globalindex.btree.btree_index_reader import BTreeIndexReader
-from pypaimon.globalindex.btree.key_serializer import KeySerializer
from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
-from pypaimon.globalindex.global_index_reader import FieldRef, GlobalIndexReader, _completed_future
-from pypaimon.globalindex.global_index_result import GlobalIndexResult
-
-
-class LazyFilteredBTreeReader(GlobalIndexReader):
- """Manages multiple BTree index files for one range.
-
- Uses BTreeFileMetaSelector to prune files before I/O, and lazily
- creates/caches BTreeIndexReader instances only for files that are
- actually visited.
- """
-
- def __init__(
- self,
- key_serializer: KeySerializer,
- file_io,
- index_path: str,
- io_metas: List[GlobalIndexIOMeta],
- executor: Executor,
- ):
- self._key_serializer = key_serializer
- self._file_io = file_io
- self._index_path = index_path
- self._executor = executor
-
- files = []
- for io_meta in io_metas:
- idx_meta = BTreeIndexMeta.deserialize(io_meta.metadata)
- files.append((io_meta, idx_meta))
-
- self._selector = BTreeFileMetaSelector(files, key_serializer)
- self._reader_cache: Dict[str, BTreeIndexReader] = {}
- self._cache_lock = threading.Lock()
-
- def _get_or_create_reader(self, meta: GlobalIndexIOMeta) -> BTreeIndexReader:
- key = meta.external_path or meta.file_name
- with self._cache_lock:
- reader = self._reader_cache.get(key)
- if reader is not None:
- return reader
- reader = BTreeIndexReader(
- key_serializer=self._key_serializer,
- file_io=self._file_io,
- index_path=self._index_path,
- io_meta=meta,
- )
- self._reader_cache[key] = reader
- return reader
-
- def _visit_parallel(
- self,
- selector_fn: Callable[[], Optional[List[GlobalIndexIOMeta]]],
- visitor_fn: Callable[[BTreeIndexReader], Optional[GlobalIndexResult]],
- ) -> 'Future[Optional[GlobalIndexResult]]':
- selected = selector_fn()
- if selected is None:
- selected = [io_meta for io_meta, _ in self._selector._files]
- if not selected:
- return _completed_future(GlobalIndexResult.create_empty())
-
- # Single-level submit: reader creation + query in one task.
- # This avoids the nested submission deadlock when using a
- # semaphore-limited executor.
- task_futures: List[Future] = []
- for meta in selected:
- task_futures.append(self._executor.submit(
- lambda m=meta: visitor_fn(self._get_or_create_reader(m))))
-
- # Union all results once every task future is done
- all_done: Future = Future()
- remaining = [len(task_futures)]
- lock = threading.Lock()
-
- def on_done(_):
- with lock:
- remaining[0] -= 1
- if remaining[0] == 0:
- try:
- result: Optional[GlobalIndexResult] = None
- for f in task_futures:
- current = f.result()
- if current is None:
- continue
- if result is None:
- result = current
- else:
- result = result.or_(current)
- all_done.set_result(result if result is not None
- else GlobalIndexResult.create_empty())
- except Exception as e:
- all_done.set_exception(e)
-
- for f in task_futures:
- f.add_done_callback(on_done)
-
- return all_done
-
- # ---- visit methods -------------------------------------------------------
-
- def visit_equal(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_equal(literal),
- lambda r: r.visit_equal(literal))
-
- def visit_not_equal(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_not_equal(literal),
- lambda r: r.visit_not_equal(literal))
-
- def visit_less_than(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_less_than(literal),
- lambda r: r.visit_less_than(literal))
-
- def visit_less_or_equal(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_less_or_equal(literal),
- lambda r: r.visit_less_or_equal(literal))
-
- def visit_greater_than(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_greater_than(literal),
- lambda r: r.visit_greater_than(literal))
-
- def visit_greater_or_equal(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_greater_or_equal(literal),
- lambda r: r.visit_greater_or_equal(literal))
-
- def visit_is_null(self, field_ref: FieldRef) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_is_null(),
- lambda r: r.visit_is_null())
-
- def visit_is_not_null(self, field_ref: FieldRef) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_is_not_null(),
- lambda r: r.visit_is_not_null())
-
- def visit_in(self, field_ref: FieldRef, literals) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_in(literals),
- lambda r: r.visit_in(literals))
-
- def visit_not_in(self, field_ref: FieldRef, literals) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_not_in(literals),
- lambda r: r.visit_not_in(literals))
-
- def visit_starts_with(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_starts_with(literal),
- lambda r: r.visit_starts_with(literal))
-
- def visit_ends_with(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_ends_with(literal),
- lambda r: r.visit_ends_with(literal))
-
- def visit_contains(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_contains(literal),
- lambda r: r.visit_contains(literal))
+from pypaimon.globalindex.sorted_file_global_index_reader import SortedFileGlobalIndexReader
- def visit_like(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_like(literal),
- lambda r: r.visit_like(literal))
- def visit_between(self, field_ref: FieldRef, from_v, to_v) -> 'Future[Optional[GlobalIndexResult]]':
- return self._visit_parallel(
- lambda: self._selector.select_between(from_v, to_v),
- lambda r: r.visit_between(from_v, to_v))
+class LazyFilteredBTreeReader(SortedFileGlobalIndexReader):
+ """Manages multiple BTree index files for one row-id range."""
- def close(self) -> None:
- with self._cache_lock:
- for reader in self._reader_cache.values():
- try:
- reader.close()
- except Exception:
- pass
- self._reader_cache.clear()
+ def open_reader(self, meta: GlobalIndexIOMeta) -> BTreeIndexReader:
+ return BTreeIndexReader(
+ key_serializer=self._key_serializer,
+ file_io=self._file_io,
+ index_path=self._index_path,
+ io_meta=meta,
+ )
diff --git a/paimon-python/pypaimon/globalindex/btree/memory_slice_input.py b/paimon-python/pypaimon/globalindex/btree/memory_slice_input.py
index fe0aee5e66..4cf97a10e7 100644
--- a/paimon-python/pypaimon/globalindex/btree/memory_slice_input.py
+++ b/paimon-python/pypaimon/globalindex/btree/memory_slice_input.py
@@ -15,147 +15,8 @@
# specific language governing permissions and limitations
# under the License.
-"""Input for byte array."""
+"""Compatibility re-export for global index byte-slice input."""
-import struct
+from pypaimon.globalindex.memory_slice_input import MemorySliceInput
-
-class MemorySliceInput:
- """Input for byte array."""
-
- def __init__(self, data: bytes):
- """Initialize MemorySliceInput.
-
- Args:
- data: The byte array to read from
- """
- self.data = data
- self._position = 0
-
- def position(self) -> int:
- """Get current position."""
- return self._position
-
- def set_position(self, position: int) -> None:
- """Set current position.
-
- Args:
- position: The new position
-
- Raises:
- IndexError: If position is out of bounds
- """
- if position < 0 or position > len(self.data):
- raise IndexError(f"Position {position} out of bounds [0, {len(self.data)}]")
- self._position = position
-
- def is_readable(self) -> bool:
- """Check if there are more bytes to read."""
- return self.available() > 0
-
- def available(self) -> int:
- """Get number of available bytes."""
- return len(self.data) - self._position
-
- def read_byte(self) -> int:
- """Read a single byte.
-
- Returns:
- The byte value (0-255)
-
- Raises:
- IndexError: If no more bytes available
- """
- if self._position >= len(self.data):
- raise IndexError("No more bytes available")
- value = self.data[self._position]
- self._position += 1
- return value
-
- def read_unsigned_byte(self) -> int:
- """Read an unsigned byte.
-
- Returns:
- The byte value (0-255)
- """
- return self.read_byte() & 0xFF
-
- def read_int(self) -> int:
- """Read a 4-byte integer (big-endian).
-
- Returns:
- The integer value
- """
- if self._position + 4 > len(self.data):
- raise IndexError("Not enough bytes to read int")
- value = struct.unpack('>I', self.data[self._position:self._position + 4])[0]
- self._position += 4
- return value
-
- def read_var_len_int(self) -> int:
- """Read a variable-length integer.
-
- Returns:
- The integer value
-
- Raises:
- ValueError: If integer is malformed
- """
- result = 0
- offset = 0
- while offset < 32:
- b = self.read_unsigned_byte()
- result |= (b & 0x7F) << offset
- if (b & 0x80) == 0:
- return result
- offset += 7
- raise ValueError("Malformed integer")
-
- def read_long(self) -> int:
- """Read an 8-byte long (big-endian).
-
- Returns:
- The long value
- """
- if self._position + 8 > len(self.data):
- raise IndexError("Not enough bytes to read long")
- value = struct.unpack('>Q', self.data[self._position:self._position + 8])[0]
- self._position += 8
- return value
-
- def read_var_len_long(self) -> int:
- """Read a variable-length long.
-
- Returns:
- The long value
-
- Raises:
- ValueError: If long is malformed
- """
- result = 0
- offset = 0
- while offset < 64:
- b = self.read_unsigned_byte()
- result |= (b & 0x7F) << offset
- if (b & 0x80) == 0:
- return result
- offset += 7
- raise ValueError("Malformed long")
-
- def read_slice(self, length: int) -> bytes:
- """Read a slice of bytes.
-
- Args:
- length: Number of bytes to read
-
- Returns:
- The slice of bytes
-
- Raises:
- IndexError: If not enough bytes available
- """
- if self._position + length > len(self.data):
- raise IndexError(f"Not enough bytes to read slice of length {length}")
- value = self.data[self._position:self._position + length]
- self._position += length
- return value
+__all__ = ['MemorySliceInput']
diff --git a/paimon-python/pypaimon/globalindex/btree/sst_file_reader.py b/paimon-python/pypaimon/globalindex/btree/sst_file_reader.py
index b2e2c87775..e766946918 100644
--- a/paimon-python/pypaimon/globalindex/btree/sst_file_reader.py
+++ b/paimon-python/pypaimon/globalindex/btree/sst_file_reader.py
@@ -27,15 +27,19 @@ Thread-safe when the underlying stream supports position-based reads
import struct
import threading
-import zlib
from typing import Optional, Callable
from typing import BinaryIO
from pypaimon.common.file_io import pread
+from pypaimon.globalindex.block_compression import (
+ BLOCK_TRAILER_LENGTH,
+ crc32c,
+ decompress_block,
+)
from pypaimon.globalindex.btree.block_handle import BlockHandle
from pypaimon.globalindex.btree.block_entry import BlockEntry
from pypaimon.globalindex.btree.block_reader import BlockReader, BlockIterator
-from pypaimon.globalindex.btree.memory_slice_input import MemorySliceInput
+from pypaimon.globalindex.memory_slice_input import MemorySliceInput
class SstFileIterator:
@@ -137,37 +141,27 @@ class SstFileReader:
def _read_block(self, block_handle: BlockHandle) -> BlockReader:
# Read block data + 5 bytes trailer (1 byte compression type + 4 bytes CRC32)
- block_data = self._read_from(block_handle.offset, block_handle.size + 5)
+ block_data = self._read_from(
+ block_handle.offset, block_handle.size + BLOCK_TRAILER_LENGTH)
# Parse block trailer (last 5 bytes: 1 byte compression type + 4 bytes CRC32)
- if len(block_data) < 5:
+ if len(block_data) < BLOCK_TRAILER_LENGTH:
raise ValueError("Block data too short to contain trailer")
- trailer_offset = len(block_data) - 5
+ trailer_offset = len(block_data) - BLOCK_TRAILER_LENGTH
compression_type = block_data[trailer_offset]
- crc32_value = struct.unpack('<I', block_data[trailer_offset + 1:trailer_offset + 5])[0]
+ crc32_value = struct.unpack(
+ '<I',
+ block_data[trailer_offset + 1:trailer_offset + BLOCK_TRAILER_LENGTH])[0]
# Extract block data (without trailer)
block_bytes = block_data[:trailer_offset]
# Verify CRC32
- actual_crc32 = self.crc32c(block_bytes, compression_type)
+ actual_crc32 = crc32c(block_bytes, compression_type)
if actual_crc32 != crc32_value:
raise ValueError(f"CRC32 mismatch: expected {crc32_value}, got {actual_crc32}")
- # Decompress if needed
- if compression_type == 1: # ZSTD
- import zstandard as zstd
- from io import BytesIO
- decompressor = zstd.ZstdDecompressor()
- memory_input = MemorySliceInput(block_bytes)
- expected_len = memory_input.read_var_len_int()
- compressed = block_bytes[memory_input.position():]
- with decompressor.stream_reader(BytesIO(compressed)) as reader:
- block_bytes = reader.read()
- if len(block_bytes) != expected_len:
- raise ValueError("Corrupted data, decompression failed.")
- elif compression_type != 0: # Not NONE
- raise ValueError(f"Compression type {compression_type} not supported")
+ block_bytes = decompress_block(block_bytes, compression_type)
return BlockReader.create(block_bytes, self.comparator)
@@ -179,27 +173,6 @@ class SstFileReader:
read_block,
self.index_block.iterator())
- @staticmethod
- def crc32c(bytes_data: bytes, compression_type_id: int) -> int:
- """
- Calculate CRC32 checksum for the given bytes and compression type.
-
- Args:
- bytes_data: The byte array to calculate checksum for
- compression_type_id: The persistent ID of the compression type (0-255)
-
- Returns:
- The CRC32 checksum value
- """
- # Calculate CRC32 for the data bytes
- crc_value = zlib.crc32(bytes_data)
-
- # Update with compression type ID (lower 8 bits only)
- crc_value = zlib.crc32(bytes([compression_type_id & 0xFF]), crc_value)
-
- # Return as unsigned 32-bit integer
- return crc_value & 0xFFFFFFFF
-
def close(self) -> None:
"""Close the reader and release resources."""
# No resources to release in this implementation
diff --git a/paimon-python/pypaimon/globalindex/global_index_evaluator.py b/paimon-python/pypaimon/globalindex/global_index_evaluator.py
index ebc05d4791..9fdf64a607 100644
--- a/paimon-python/pypaimon/globalindex/global_index_evaluator.py
+++ b/paimon-python/pypaimon/globalindex/global_index_evaluator.py
@@ -222,6 +222,8 @@ class GlobalIndexEvaluator:
return reader.visit_like(field_ref, literals[0])
elif method == 'between':
return reader.visit_between(field_ref, literals[0], literals[1])
+ elif method == 'notBetween':
+ return reader.visit_not_between(field_ref, literals[0], literals[1])
from pypaimon.globalindex.global_index_reader import _completed_future
return _completed_future(None)
diff --git a/paimon-python/pypaimon/globalindex/global_index_reader.py b/paimon-python/pypaimon/globalindex/global_index_reader.py
index 84825c7056..1ac0905bb6 100644
--- a/paimon-python/pypaimon/globalindex/global_index_reader.py
+++ b/paimon-python/pypaimon/globalindex/global_index_reader.py
@@ -106,6 +106,11 @@ class GlobalIndexReader(ABC):
def visit_between(self, field_ref: FieldRef, min_v: object, max_v: object) -> 'Future[Optional[GlobalIndexResult]]':
return _completed_future(None)
+ def visit_not_between(
+ self, field_ref: FieldRef, min_v: object, max_v: object
+ ) -> 'Future[Optional[GlobalIndexResult]]':
+ return _completed_future(None)
+
@abstractmethod
def close(self) -> None:
pass
diff --git a/paimon-python/pypaimon/globalindex/global_index_scanner.py b/paimon-python/pypaimon/globalindex/global_index_scanner.py
index 924d30608e..23af17e405 100644
--- a/paimon-python/pypaimon/globalindex/global_index_scanner.py
+++ b/paimon-python/pypaimon/globalindex/global_index_scanner.py
@@ -24,6 +24,8 @@ from pypaimon.globalindex.global_index_evaluator import GlobalIndexEvaluator
from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
from pypaimon.globalindex.global_index_reader import GlobalIndexReader
from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.options import Options
from pypaimon.common.predicate import Predicate
from pypaimon.read.push_down_utils import _get_all_fields
from pypaimon.schema.data_types import DataField
@@ -40,7 +42,9 @@ class GlobalIndexScanner:
index_path: str,
index_files: Collection['IndexFileMeta'],
thread_num: Optional[int] = None,
+ options: Optional[CoreOptions] = None,
):
+ self._options = options or CoreOptions(Options.from_none())
self._executor = ThreadPoolExecutor(
max_workers=thread_num or 32
)
@@ -76,9 +80,11 @@ class GlobalIndexScanner:
index_metas[field_id][index_type][range_key].append(io_meta)
executor = self._executor
+ options = self._options
def readers_function(field: DataField) -> Collection[GlobalIndexReader]:
- return _create_readers(file_io, index_path, index_metas.get(field.id), field, executor)
+ return _create_readers(
+ file_io, index_path, index_metas.get(field.id), field, executor, options)
return GlobalIndexEvaluator(fields, readers_function)
@@ -105,6 +111,7 @@ class GlobalIndexScanner:
index_path=table.path_factory().global_index_path_factory().index_path(),
index_files=index_files,
thread_num=table.options.global_index_thread_num(),
+ options=table.options,
)
# Scan index files from snapshot using partition_filter and predicate
@@ -138,6 +145,7 @@ class GlobalIndexScanner:
index_path=table.path_factory().global_index_path_factory().index_path(),
index_files=scanned_index_files,
thread_num=table.options.global_index_thread_num(),
+ options=table.options,
)
def scan(self, predicate: Optional[Predicate]) -> Optional[GlobalIndexResult]:
@@ -156,7 +164,7 @@ class GlobalIndexScanner:
self.close()
-def _create_readers(file_io, index_path, index_type_metas, field, executor=None):
+def _create_readers(file_io, index_path, index_type_metas, field, executor=None, options=None):
"""Create readers for a specific field, dispatched by index_type.
Unknown indexTypes raise — a silent skip would make
@@ -178,7 +186,7 @@ def _create_readers(file_io, index_path, index_type_metas, field, executor=None)
offset_readers = []
for range_key, io_metas in range_metas.items():
inner_readers = _create_inner_readers(
- index_type, file_io, index_path, field, io_metas, executor)
+ index_type, file_io, index_path, field, io_metas, executor, options)
for inner in inner_readers:
offset_readers.append(
OffsetGlobalIndexReader(
@@ -188,11 +196,13 @@ def _create_readers(file_io, index_path, index_type_metas, field, executor=None)
return readers
-def _create_inner_readers(index_type, file_io, index_path, field, io_metas, executor=None):
+def _create_inner_readers(
+ index_type, file_io, index_path, field, io_metas, executor=None, options=None):
"""Build per-file (or per-shard) readers for a single indexType/range."""
+ core_options = options or CoreOptions(Options.from_none())
if index_type == 'btree':
from pypaimon.globalindex.btree.lazy_filtered_btree_reader import LazyFilteredBTreeReader
- from pypaimon.globalindex.btree.key_serializer import create_serializer
+ from pypaimon.globalindex.key_serializer import create_serializer
key_serializer = create_serializer(field.type)
return [LazyFilteredBTreeReader(
key_serializer=key_serializer,
@@ -200,6 +210,20 @@ def _create_inner_readers(index_type, file_io, index_path, field, io_metas, exec
index_path=index_path,
io_metas=io_metas,
executor=executor,
+ fallback_scan_max_size=core_options.btree_index_fallback_scan_max_size(),
+ )]
+
+ if index_type == 'bitmap':
+ from pypaimon.globalindex.bitmap.lazy_filtered_bitmap_reader import LazyFilteredBitmapReader
+ from pypaimon.globalindex.key_serializer import create_serializer
+ key_serializer = create_serializer(field.type)
+ return [LazyFilteredBitmapReader(
+ key_serializer=key_serializer,
+ file_io=file_io,
+ index_path=index_path,
+ io_metas=io_metas,
+ executor=executor,
+ fallback_scan_max_size=core_options.bitmap_index_fallback_scan_max_size(),
)]
from pypaimon.globalindex.tantivy import (
diff --git a/paimon-python/pypaimon/globalindex/btree/key_serializer.py b/paimon-python/pypaimon/globalindex/key_serializer.py
similarity index 73%
copy from paimon-python/pypaimon/globalindex/btree/key_serializer.py
copy to paimon-python/pypaimon/globalindex/key_serializer.py
index 58f35bab25..875d6b0bb1 100644
--- a/paimon-python/pypaimon/globalindex/btree/key_serializer.py
+++ b/paimon-python/pypaimon/globalindex/key_serializer.py
@@ -15,22 +15,18 @@
# specific language governing permissions and limitations
# under the License.
-"""Key serializer for B-tree index."""
+"""Key serializer for global indexes."""
from abc import ABC, abstractmethod
from typing import Callable
import struct
-from pypaimon.schema.data_types import DataType
from pypaimon.schema.data_types import AtomicType
+from pypaimon.schema.data_types import DataType
class KeySerializer(ABC):
- """
- Interface for serializing and deserializing B-tree index keys.
-
- This interface provides core methods to ser/de and compare btree index keys.
- """
+ """Interface for serializing, deserializing, and comparing global index keys."""
@abstractmethod
def serialize(self, key: object) -> bytes:
@@ -44,15 +40,7 @@ class KeySerializer(ABC):
@abstractmethod
def create_comparator(self) -> Callable[[object, object], int]:
- """
- Create a comparator function for keys.
-
- Returns:
- A function that takes two keys and returns:
- - negative if first < second
- - 0 if first == second
- - positive if first > second
- """
+ """Create a comparator function for keys."""
pass
@@ -60,23 +48,20 @@ class StringSerializer(KeySerializer):
"""Serializer for STRING type."""
def serialize(self, key: object) -> bytes:
- """Serialize a string key to bytes."""
if isinstance(key, str):
return key.encode('utf-8')
return str(key).encode('utf-8')
def deserialize(self, data: bytes) -> object:
- """Deserialize bytes to a string key."""
return data.decode('utf-8')
def create_comparator(self) -> Callable[[object, object], int]:
- """Create a comparator for string keys."""
def compare(a: object, b: object) -> int:
str_a = a if isinstance(a, str) else str(a)
str_b = b if isinstance(b, str) else str(b)
if str_a < str_b:
return -1
- elif str_a > str_b:
+ if str_a > str_b:
return 1
return 0
return compare
@@ -86,21 +71,18 @@ class LongSerializer(KeySerializer):
"""Serializer for BIGINT type."""
def serialize(self, key: object) -> bytes:
- """Serialize a long key to bytes."""
return struct.pack('<q', int(key))
def deserialize(self, data: bytes) -> object:
- """Deserialize bytes to a long key."""
return struct.unpack('<q', data)[0]
def create_comparator(self) -> Callable[[object, object], int]:
- """Create a comparator for long keys."""
def compare(a: object, b: object) -> int:
long_a = int(a)
long_b = int(b)
if long_a < long_b:
return -1
- elif long_a > long_b:
+ if long_a > long_b:
return 1
return 0
return compare
@@ -110,21 +92,18 @@ class IntSerializer(KeySerializer):
"""Serializer for INT type."""
def serialize(self, key: object) -> bytes:
- """Serialize an int key to bytes."""
return struct.pack('<i', int(key))
def deserialize(self, data: bytes) -> object:
- """Deserialize bytes to an int key."""
return struct.unpack('<i', data)[0]
def create_comparator(self) -> Callable[[object, object], int]:
- """Create a comparator for int keys."""
def compare(a: object, b: object) -> int:
int_a = int(a)
int_b = int(b)
if int_a < int_b:
return -1
- elif int_a > int_b:
+ if int_a > int_b:
return 1
return 0
return compare
@@ -132,13 +111,13 @@ class IntSerializer(KeySerializer):
def create_serializer(data_type: DataType) -> KeySerializer:
if not isinstance(data_type, AtomicType):
- raise ValueError(f"Key serializer only support AtomicType yet, meet {data_type.__class__}")
+ raise ValueError(
+ f"Key serializer only support AtomicType yet, meet {data_type.__class__}")
type_name = data_type.type.upper()
if type_name in ('CHAR', 'VARCHAR', 'STRING'):
return StringSerializer()
- elif type_name == 'BIGINT':
+ if type_name == 'BIGINT':
return LongSerializer()
- elif type_name == 'INT':
+ if type_name == 'INT':
return IntSerializer()
- else:
- raise ValueError(f"DataType: {data_type} is not supported by btree index now.")
+ raise ValueError(f"DataType: {data_type} is not supported by global index now.")
diff --git a/paimon-python/pypaimon/globalindex/btree/memory_slice_input.py b/paimon-python/pypaimon/globalindex/memory_slice_input.py
similarity index 68%
copy from paimon-python/pypaimon/globalindex/btree/memory_slice_input.py
copy to paimon-python/pypaimon/globalindex/memory_slice_input.py
index fe0aee5e66..e8e81b7fa4 100644
--- a/paimon-python/pypaimon/globalindex/btree/memory_slice_input.py
+++ b/paimon-python/pypaimon/globalindex/memory_slice_input.py
@@ -22,85 +22,51 @@ import struct
class MemorySliceInput:
"""Input for byte array."""
-
+
def __init__(self, data: bytes):
- """Initialize MemorySliceInput.
-
- Args:
- data: The byte array to read from
- """
self.data = data
self._position = 0
-
+
def position(self) -> int:
"""Get current position."""
return self._position
-
+
def set_position(self, position: int) -> None:
- """Set current position.
-
- Args:
- position: The new position
-
- Raises:
- IndexError: If position is out of bounds
- """
+ """Set current position."""
if position < 0 or position > len(self.data):
raise IndexError(f"Position {position} out of bounds [0, {len(self.data)}]")
self._position = position
-
+
def is_readable(self) -> bool:
"""Check if there are more bytes to read."""
return self.available() > 0
-
+
def available(self) -> int:
"""Get number of available bytes."""
return len(self.data) - self._position
-
+
def read_byte(self) -> int:
- """Read a single byte.
-
- Returns:
- The byte value (0-255)
-
- Raises:
- IndexError: If no more bytes available
- """
+ """Read a single byte."""
if self._position >= len(self.data):
raise IndexError("No more bytes available")
value = self.data[self._position]
self._position += 1
return value
-
+
def read_unsigned_byte(self) -> int:
- """Read an unsigned byte.
-
- Returns:
- The byte value (0-255)
- """
+ """Read an unsigned byte."""
return self.read_byte() & 0xFF
-
+
def read_int(self) -> int:
- """Read a 4-byte integer (big-endian).
-
- Returns:
- The integer value
- """
+ """Read a 4-byte integer (big-endian)."""
if self._position + 4 > len(self.data):
raise IndexError("Not enough bytes to read int")
value = struct.unpack('>I', self.data[self._position:self._position + 4])[0]
self._position += 4
return value
-
+
def read_var_len_int(self) -> int:
- """Read a variable-length integer.
-
- Returns:
- The integer value
-
- Raises:
- ValueError: If integer is malformed
- """
+ """Read a variable-length integer."""
result = 0
offset = 0
while offset < 32:
@@ -110,28 +76,17 @@ class MemorySliceInput:
return result
offset += 7
raise ValueError("Malformed integer")
-
+
def read_long(self) -> int:
- """Read an 8-byte long (big-endian).
-
- Returns:
- The long value
- """
+ """Read an 8-byte long (big-endian)."""
if self._position + 8 > len(self.data):
raise IndexError("Not enough bytes to read long")
value = struct.unpack('>Q', self.data[self._position:self._position + 8])[0]
self._position += 8
return value
-
+
def read_var_len_long(self) -> int:
- """Read a variable-length long.
-
- Returns:
- The long value
-
- Raises:
- ValueError: If long is malformed
- """
+ """Read a variable-length long."""
result = 0
offset = 0
while offset < 64:
@@ -141,19 +96,9 @@ class MemorySliceInput:
return result
offset += 7
raise ValueError("Malformed long")
-
+
def read_slice(self, length: int) -> bytes:
- """Read a slice of bytes.
-
- Args:
- length: Number of bytes to read
-
- Returns:
- The slice of bytes
-
- Raises:
- IndexError: If not enough bytes available
- """
+ """Read a slice of bytes."""
if self._position + length > len(self.data):
raise IndexError(f"Not enough bytes to read slice of length {length}")
value = self.data[self._position:self._position + length]
diff --git a/paimon-python/pypaimon/globalindex/offset_global_index_reader.py b/paimon-python/pypaimon/globalindex/offset_global_index_reader.py
index 8d2c5bc38b..a0965ed205 100644
--- a/paimon-python/pypaimon/globalindex/offset_global_index_reader.py
+++ b/paimon-python/pypaimon/globalindex/offset_global_index_reader.py
@@ -97,5 +97,10 @@ class OffsetGlobalIndexReader(GlobalIndexReader):
def visit_between(self, field_ref: FieldRef, min_v: object, max_v: object) -> 'Future[Optional[GlobalIndexResult]]':
return self._apply_offset_future(self._wrapped.visit_between(field_ref, min_v, max_v))
+ def visit_not_between(
+ self, field_ref: FieldRef, min_v: object, max_v: object
+ ) -> 'Future[Optional[GlobalIndexResult]]':
+ return self._apply_offset_future(self._wrapped.visit_not_between(field_ref, min_v, max_v))
+
def close(self) -> None:
self._wrapped.close()
diff --git a/paimon-python/pypaimon/globalindex/sorted_file_global_index_reader.py b/paimon-python/pypaimon/globalindex/sorted_file_global_index_reader.py
new file mode 100644
index 0000000000..273a701c74
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/sorted_file_global_index_reader.py
@@ -0,0 +1,310 @@
+# 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.
+
+"""Base reader for sorted global index files with manifest pruning."""
+
+import threading
+from concurrent.futures import Executor, Future
+from typing import Callable, Dict, List, Optional
+
+from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
+from pypaimon.globalindex.global_index_reader import FieldRef, GlobalIndexReader, _completed_future
+from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.globalindex.key_serializer import KeySerializer
+from pypaimon.globalindex.sorted_file_meta_selector import SortedFileMetaSelector
+from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
+from pypaimon.utils.roaring_bitmap import RoaringBitmap64
+
+
+_LONG_MAX_VALUE = (1 << 63) - 1
+
+
+class SortedFileGlobalIndexReader(GlobalIndexReader):
+ """Shared lazy reader/candidate pruning logic for sorted global indexes."""
+
+ def __init__(
+ self,
+ key_serializer: KeySerializer,
+ file_io,
+ index_path: str,
+ io_metas: List[GlobalIndexIOMeta],
+ executor: Executor,
+ fallback_scan_max_size: int,
+ ):
+ self._key_serializer = key_serializer
+ self._file_io = file_io
+ self._index_path = index_path
+ self._executor = executor
+ self._fallback_scan_max_size = fallback_scan_max_size
+
+ files = []
+ for io_meta in io_metas:
+ idx_meta = SortedIndexFileMeta.deserialize(io_meta.metadata)
+ files.append((io_meta, idx_meta))
+
+ self._selector = SortedFileMetaSelector(files, key_serializer)
+ self._reader_cache: Dict[str, object] = {}
+ self._cache_lock = threading.Lock()
+
+ def open_reader(self, meta: GlobalIndexIOMeta):
+ raise NotImplementedError
+
+ def _get_or_create_reader(self, meta: GlobalIndexIOMeta):
+ key = meta.external_path or meta.file_name
+ with self._cache_lock:
+ reader = self._reader_cache.get(key)
+ if reader is not None:
+ return reader
+ reader = self.open_reader(meta)
+ self._reader_cache[key] = reader
+ return reader
+
+ def _visit_parallel(
+ self,
+ selector_fn: Callable[[], Optional[List[GlobalIndexIOMeta]]],
+ visitor_fn: Callable[[object], Optional[GlobalIndexResult]],
+ ) -> 'Future[Optional[GlobalIndexResult]]':
+ selected = selector_fn()
+ if selected is None:
+ return _completed_future(None)
+ if not selected:
+ return _completed_future(GlobalIndexResult.create_empty())
+
+ task_futures: List[Future] = []
+ for meta in selected:
+ task_futures.append(self._executor.submit(
+ lambda m=meta: visitor_fn(self._get_or_create_reader(m))))
+
+ all_done: Future = Future()
+ remaining = [len(task_futures)]
+ lock = threading.Lock()
+
+ def on_done(_):
+ with lock:
+ remaining[0] -= 1
+ if remaining[0] == 0:
+ try:
+ result: Optional[GlobalIndexResult] = None
+ for future in task_futures:
+ current = future.result()
+ if current is None:
+ continue
+ if result is None:
+ result = current
+ else:
+ result = result.or_(current)
+ all_done.set_result(result)
+ except Exception as e:
+ all_done.set_exception(e)
+
+ for future in task_futures:
+ future.add_done_callback(on_done)
+
+ return all_done
+
+ def _visit_fallback_parallel(
+ self,
+ selector_fn: Callable[[], Optional[List[GlobalIndexIOMeta]]],
+ visitor_fn: Callable[[object], Optional[GlobalIndexResult]],
+ ) -> 'Future[Optional[GlobalIndexResult]]':
+ selected = selector_fn()
+ if selected is None:
+ return _completed_future(None)
+ if not selected:
+ return _completed_future(GlobalIndexResult.create_empty())
+ if not self._fallback_scan_enabled(selected):
+ return _completed_future(None)
+ return self._visit_parallel(lambda: selected, visitor_fn)
+
+ def _fallback_scan_enabled(self, files: List[GlobalIndexIOMeta]) -> bool:
+ if self._fallback_scan_max_size <= 0:
+ return False
+ total_size = 0
+ for file in files:
+ if _LONG_MAX_VALUE - total_size < file.file_size:
+ return False
+ total_size += file.file_size
+ if total_size > self._fallback_scan_max_size:
+ return False
+ return True
+
+ def visit_equal(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
+ return self._visit_parallel(
+ lambda: self._selector.select_equal(literal),
+ lambda reader: reader.visit_equal(literal))
+
+ def visit_not_equal(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
+ return self._visit_parallel(
+ lambda: self._selector.select_not_equal(literal),
+ lambda reader: reader.visit_not_equal(literal))
+
+ def visit_less_than(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
+ if not self._can_fallback_scan(literal):
+ return _completed_future(None)
+ return self._visit_fallback_parallel(
+ lambda: self._selector.select_less_than(literal),
+ lambda reader: reader.visit_less_than(literal))
+
+ def visit_less_or_equal(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
+ if not self._can_fallback_scan(literal):
+ return _completed_future(None)
+ return self._visit_fallback_parallel(
+ lambda: self._selector.select_less_or_equal(literal),
+ lambda reader: reader.visit_less_or_equal(literal))
+
+ def visit_greater_than(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
+ if not self._can_fallback_scan(literal):
+ return _completed_future(None)
+ return self._visit_fallback_parallel(
+ lambda: self._selector.select_greater_than(literal),
+ lambda reader: reader.visit_greater_than(literal))
+
+ def visit_greater_or_equal(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
+ if not self._can_fallback_scan(literal):
+ return _completed_future(None)
+ return self._visit_fallback_parallel(
+ lambda: self._selector.select_greater_or_equal(literal),
+ lambda reader: reader.visit_greater_or_equal(literal))
+
+ def visit_is_null(self, field_ref: FieldRef) -> 'Future[Optional[GlobalIndexResult]]':
+ return self._visit_parallel(
+ lambda: self._selector.select_is_null(),
+ lambda reader: reader.visit_is_null())
+
+ def visit_is_not_null(self, field_ref: FieldRef) -> 'Future[Optional[GlobalIndexResult]]':
+ return self._visit_parallel(
+ lambda: self._selector.select_is_not_null(),
+ lambda reader: reader.visit_is_not_null())
+
+ def visit_in(self, field_ref: FieldRef, literals) -> 'Future[Optional[GlobalIndexResult]]':
+ return self._visit_parallel(
+ lambda: self._selector.select_in(literals),
+ lambda reader: reader.visit_in(literals))
+
+ def visit_not_in(self, field_ref: FieldRef, literals) -> 'Future[Optional[GlobalIndexResult]]':
+ return self._visit_parallel(
+ lambda: self._selector.select_not_in(literals),
+ lambda reader: reader.visit_not_in(literals))
+
+ def visit_starts_with(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
+ if not self._is_string_type(field_ref) or literal is None:
+ return _completed_future(None)
+ return self._visit_parallel(
+ lambda: self._selector.select_starts_with(literal),
+ lambda reader: reader.visit_starts_with(literal))
+
+ def visit_ends_with(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
+ if not self._can_fallback_string_scan(field_ref, literal):
+ return _completed_future(None)
+ return self._visit_fallback_parallel(
+ lambda: self._selector.select_ends_with(literal),
+ lambda reader: reader.visit_ends_with(literal))
+
+ def visit_contains(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
+ if not self._can_fallback_string_scan(field_ref, literal):
+ return _completed_future(None)
+ return self._visit_fallback_parallel(
+ lambda: self._selector.select_contains(literal),
+ lambda reader: reader.visit_contains(literal))
+
+ def visit_like(self, field_ref: FieldRef, literal) -> 'Future[Optional[GlobalIndexResult]]':
+ if not self._is_string_type(field_ref) or literal is None:
+ return _completed_future(None)
+
+ optimized = self._try_optimize_like(literal)
+ if optimized is not None:
+ method, optimized_literal = optimized
+ if method == "equal":
+ return self.visit_equal(field_ref, optimized_literal)
+ if method == "starts_with":
+ return self.visit_starts_with(field_ref, optimized_literal)
+ if method == "ends_with":
+ return self.visit_ends_with(field_ref, optimized_literal)
+ if method == "contains":
+ return self.visit_contains(field_ref, optimized_literal)
+
+ if not self._can_fallback_string_scan(field_ref, literal):
+ return _completed_future(None)
+ return self._visit_fallback_parallel(
+ lambda: self._selector.select_like(literal),
+ lambda reader: reader.visit_like(literal))
+
+ def visit_between(self, field_ref: FieldRef, from_v, to_v) -> 'Future[Optional[GlobalIndexResult]]':
+ if not self._can_fallback_scan(from_v) or to_v is None:
+ return _completed_future(None)
+ return self._visit_fallback_parallel(
+ lambda: self._selector.select_between(from_v, to_v),
+ lambda reader: reader.visit_between(from_v, to_v))
+
+ def visit_not_between(self, field_ref: FieldRef, from_v, to_v) -> 'Future[Optional[GlobalIndexResult]]':
+ if not self._can_fallback_scan(from_v) or to_v is None:
+ return _completed_future(None)
+ return self._visit_fallback_parallel(
+ lambda: self._selector.select_or([
+ self._selector.select_less_than(from_v),
+ self._selector.select_greater_than(to_v),
+ ]),
+ lambda reader: GlobalIndexResult.create(
+ RoaringBitmap64.or_(
+ reader.less_than(from_v),
+ reader.greater_than(to_v))))
+
+ def _can_fallback_string_scan(self, field_ref: FieldRef, literal) -> bool:
+ return (self._fallback_scan_max_size > 0
+ and literal is not None
+ and self._is_string_type(field_ref))
+
+ def _can_fallback_scan(self, literal) -> bool:
+ return self._fallback_scan_max_size > 0 and literal is not None
+
+ @staticmethod
+ def _is_string_type(field_ref: FieldRef) -> bool:
+ data_type = str(getattr(field_ref, "data_type", "")).upper()
+ if data_type.endswith(" NOT NULL"):
+ data_type = data_type[:-9].rstrip()
+ return (data_type == "STRING"
+ or data_type == "CHAR"
+ or data_type.startswith("CHAR(")
+ or data_type == "VARCHAR"
+ or data_type.startswith("VARCHAR("))
+
+ @staticmethod
+ def _try_optimize_like(literal):
+ pattern = str(literal)
+ if "_" in pattern:
+ return None
+ if "%" not in pattern and pattern:
+ return "equal", pattern
+ if (pattern.startswith("%")
+ and pattern.endswith("%")
+ and pattern.count("%") == 2
+ and pattern[1:-1]):
+ return "contains", pattern[1:-1]
+ if pattern.startswith("%") and pattern.count("%") == 1 and pattern[1:]:
+ return "ends_with", pattern[1:]
+ if pattern.endswith("%") and pattern.count("%") == 1 and pattern[:-1]:
+ return "starts_with", pattern[:-1]
+ return None
+
+ def close(self) -> None:
+ with self._cache_lock:
+ for reader in self._reader_cache.values():
+ try:
+ reader.close()
+ except Exception:
+ pass
+ self._reader_cache.clear()
diff --git a/paimon-python/pypaimon/globalindex/sorted_file_meta_selector.py b/paimon-python/pypaimon/globalindex/sorted_file_meta_selector.py
new file mode 100644
index 0000000000..f486d5900f
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/sorted_file_meta_selector.py
@@ -0,0 +1,169 @@
+# 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.
+
+"""Prunes sorted global index files by manifest min/max metadata."""
+
+from typing import List, Optional, Tuple
+
+from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
+from pypaimon.globalindex.key_serializer import KeySerializer
+from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
+
+
+class SortedFileMetaSelector:
+ """Selects candidate sorted global index files for a predicate."""
+
+ def __init__(
+ self,
+ files: List[Tuple[GlobalIndexIOMeta, SortedIndexFileMeta]],
+ key_serializer: KeySerializer,
+ ):
+ self._files = files
+ self._key_serializer = key_serializer
+ self._comparator = key_serializer.create_comparator()
+
+ def _filter(self, predicate) -> List[GlobalIndexIOMeta]:
+ return [io_meta for io_meta, idx_meta in self._files if predicate(idx_meta)]
+
+ def select_is_null(self) -> List[GlobalIndexIOMeta]:
+ return self._filter(lambda meta: meta.has_nulls)
+
+ def select_is_not_null(self) -> List[GlobalIndexIOMeta]:
+ return self._filter(lambda meta: not meta.only_nulls())
+
+ def select_starts_with(self, literal) -> List[GlobalIndexIOMeta]:
+ if literal is None:
+ return []
+
+ prefix = self._key_serializer.serialize(literal)
+ if len(prefix) == 0:
+ return self._filter(lambda meta: not meta.only_nulls())
+
+ prefix_key = self._key_serializer.deserialize(prefix)
+ upper_bound_bytes = self._prefix_upper_bound(prefix)
+ upper_bound = (None if upper_bound_bytes is None
+ else self._key_serializer.deserialize(upper_bound_bytes))
+
+ return self._filter(
+ lambda meta: (
+ not meta.only_nulls()
+ and self._compare_last_key(meta, prefix_key) >= 0
+ and (upper_bound is None
+ or self._compare_first_key(meta, upper_bound) < 0)))
+
+ def select_ends_with(self, literal) -> List[GlobalIndexIOMeta]:
+ return self._filter(lambda meta: literal is not None and not meta.only_nulls())
+
+ def select_contains(self, literal) -> List[GlobalIndexIOMeta]:
+ return self._filter(lambda meta: literal is not None and not meta.only_nulls())
+
+ def select_like(self, literal) -> List[GlobalIndexIOMeta]:
+ return self._filter(lambda meta: literal is not None and not meta.only_nulls())
+
+ def select_less_than(self, literal) -> List[GlobalIndexIOMeta]:
+ if literal is None:
+ return []
+ return self._filter(
+ lambda meta: not meta.only_nulls()
+ and self._compare_first_key(meta, literal) < 0)
+
+ def select_greater_or_equal(self, literal) -> List[GlobalIndexIOMeta]:
+ if literal is None:
+ return []
+ return self._filter(
+ lambda meta: not meta.only_nulls()
+ and self._compare_last_key(meta, literal) >= 0)
+
+ def select_not_equal(self, literal) -> List[GlobalIndexIOMeta]:
+ if literal is None:
+ return []
+ return self._filter(lambda meta: not meta.only_nulls())
+
+ def select_less_or_equal(self, literal) -> List[GlobalIndexIOMeta]:
+ if literal is None:
+ return []
+ return self._filter(
+ lambda meta: not meta.only_nulls()
+ and self._compare_first_key(meta, literal) <= 0)
+
+ def select_equal(self, literal) -> List[GlobalIndexIOMeta]:
+ if literal is None:
+ return []
+ return self._filter(
+ lambda meta: not meta.only_nulls()
+ and self._overlaps(meta, literal, literal))
+
+ def select_greater_than(self, literal) -> List[GlobalIndexIOMeta]:
+ if literal is None:
+ return []
+ return self._filter(
+ lambda meta: not meta.only_nulls()
+ and self._compare_last_key(meta, literal) > 0)
+
+ def select_in(self, literals) -> List[GlobalIndexIOMeta]:
+ return self._filter(lambda meta: self._matches_any_literal(meta, literals))
+
+ def select_not_in(self, literals) -> List[GlobalIndexIOMeta]:
+ if any(literal is None for literal in literals):
+ return []
+ return self._filter(lambda meta: not meta.only_nulls())
+
+ def select_between(self, from_v, to_v) -> List[GlobalIndexIOMeta]:
+ if from_v is None or to_v is None or self._comparator(from_v, to_v) > 0:
+ return []
+ return self._filter(
+ lambda meta: not meta.only_nulls()
+ and self._overlaps(meta, from_v, to_v))
+
+ def select_or(self, children: List[Optional[List[GlobalIndexIOMeta]]]
+ ) -> Optional[List[GlobalIndexIOMeta]]:
+ result = {}
+ for child in children:
+ if child is None:
+ return None
+ for io_meta in child:
+ key = io_meta.external_path or io_meta.file_name
+ result[key] = io_meta
+ return list(result.values())
+
+ def _matches_any_literal(self, meta: SortedIndexFileMeta, literals) -> bool:
+ if meta.only_nulls():
+ return False
+ for literal in literals:
+ if literal is not None and self._overlaps(meta, literal, literal):
+ return True
+ return False
+
+ def _overlaps(self, meta: SortedIndexFileMeta, from_v, to_v) -> bool:
+ return (self._comparator(from_v, self._key_serializer.deserialize(meta.last_key)) <= 0
+ and self._comparator(to_v, self._key_serializer.deserialize(meta.first_key)) >= 0)
+
+ def _compare_first_key(self, meta: SortedIndexFileMeta, literal) -> int:
+ return self._comparator(self._key_serializer.deserialize(meta.first_key), literal)
+
+ def _compare_last_key(self, meta: SortedIndexFileMeta, literal) -> int:
+ return self._comparator(self._key_serializer.deserialize(meta.last_key), literal)
+
+ @staticmethod
+ def _prefix_upper_bound(prefix: bytes) -> Optional[bytes]:
+ for index in range(len(prefix) - 1, -1, -1):
+ unsigned_byte = prefix[index]
+ if unsigned_byte != 0xFF:
+ upper_bound = bytearray(prefix[:index + 1])
+ upper_bound[index] = unsigned_byte + 1
+ return bytes(upper_bound)
+ return None
diff --git a/paimon-python/pypaimon/globalindex/btree/btree_index_meta.py b/paimon-python/pypaimon/globalindex/sorted_index_file_meta.py
similarity index 57%
copy from paimon-python/pypaimon/globalindex/btree/btree_index_meta.py
copy to paimon-python/pypaimon/globalindex/sorted_index_file_meta.py
index ce12146c20..be3e10b73a 100644
--- a/paimon-python/pypaimon/globalindex/btree/btree_index_meta.py
+++ b/paimon-python/pypaimon/globalindex/sorted_index_file_meta.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-"""B-tree index metadata."""
+"""Manifest-level min/max metadata for one sorted global index file."""
from dataclasses import dataclass
from typing import Optional
@@ -23,40 +23,48 @@ import struct
@dataclass
-class BTreeIndexMeta:
- """
- Index Meta of each BTree index file.
-
- The first key and last key of this meta could be null if the
- entire btree index file only contains nulls.
- """
+class SortedIndexFileMeta:
+ """Metadata shared by sorted global indexes such as btree and bitmap."""
+
+ FORMAT_VERSION_WITH_NULL_FLAGS = 1
+ FIRST_KEY_IS_NULL = 1
+ LAST_KEY_IS_NULL = 1 << 1
first_key: Optional[bytes]
last_key: Optional[bytes]
has_nulls: bool
def only_nulls(self) -> bool:
- """Check if this index only contains nulls."""
return self.first_key is None and self.last_key is None
@classmethod
- def deserialize(cls, data: bytes) -> 'BTreeIndexMeta':
- """Deserialize metadata from byte array."""
+ def deserialize(cls, data: bytes) -> 'SortedIndexFileMeta':
offset = 0
-
- # Read first key
+
first_key_length = struct.unpack('<I', data[offset:offset + 4])[0]
offset += 4
- first_key = None if first_key_length == 0 else data[offset:offset + first_key_length]
+ first_key = data[offset:offset + first_key_length]
offset += first_key_length
-
- # Read last key
+
last_key_length = struct.unpack('<I', data[offset:offset + 4])[0]
offset += 4
- last_key = None if last_key_length == 0 else data[offset:offset + last_key_length]
+ last_key = data[offset:offset + last_key_length]
offset += last_key_length
-
- # Read has_nulls flag
+
has_nulls = struct.unpack('<B', data[offset:offset + 1])[0] == 1
-
+ offset += 1
+
+ if len(data) - offset >= 2:
+ format_version = struct.unpack('<B', data[offset:offset + 1])[0]
+ offset += 1
+ if format_version == cls.FORMAT_VERSION_WITH_NULL_FLAGS:
+ null_key_flags = struct.unpack('<B', data[offset:offset + 1])[0]
+ if null_key_flags & cls.FIRST_KEY_IS_NULL:
+ first_key = None
+ if null_key_flags & cls.LAST_KEY_IS_NULL:
+ last_key = None
+ elif first_key_length == 0 and last_key_length == 0 and has_nulls:
+ first_key = None
+ last_key = None
+
return cls(first_key, last_key, has_nulls)
diff --git a/paimon-python/pypaimon/globalindex/union_global_index_reader.py b/paimon-python/pypaimon/globalindex/union_global_index_reader.py
index 6e2e2a9f9c..40d5e57386 100644
--- a/paimon-python/pypaimon/globalindex/union_global_index_reader.py
+++ b/paimon-python/pypaimon/globalindex/union_global_index_reader.py
@@ -123,6 +123,9 @@ class UnionGlobalIndexReader(GlobalIndexReader):
def visit_between(self, field_ref: FieldRef, from_v, to_v) -> 'Future[Optional[GlobalIndexResult]]':
return self._union_futures(lambda r: r.visit_between(field_ref, from_v, to_v))
+ def visit_not_between(self, field_ref: FieldRef, from_v, to_v) -> 'Future[Optional[GlobalIndexResult]]':
+ return self._union_futures(lambda r: r.visit_not_between(field_ref, from_v, to_v))
+
def close(self) -> None:
for reader in self._readers:
try:
diff --git a/paimon-python/pypaimon/tests/btree_thread_safety_test.py b/paimon-python/pypaimon/tests/btree_thread_safety_test.py
index b328b512b5..55bc96c5d2 100644
--- a/paimon-python/pypaimon/tests/btree_thread_safety_test.py
+++ b/paimon-python/pypaimon/tests/btree_thread_safety_test.py
@@ -42,10 +42,10 @@ def _int_meta(first: int, last: int) -> bytes:
struct.pack('<B', has_nulls))
-def _io_meta(file_name: str, first: int, last: int) -> GlobalIndexIOMeta:
+def _io_meta(file_name: str, first: int, last: int, file_size: int = 1024) -> GlobalIndexIOMeta:
return GlobalIndexIOMeta(
file_name=file_name,
- file_size=1024,
+ file_size=file_size,
metadata=_int_meta(first, last),
)
@@ -69,6 +69,12 @@ class _SlowBTreeReader:
bm.add_range(0, 10)
return GlobalIndexResult.create(bm)
+ def visit_starts_with(self, literal):
+ time.sleep(0.01)
+ bm = RoaringBitmap64()
+ bm.add_range(0, 10)
+ return GlobalIndexResult.create(bm)
+
def close(self):
pass
@@ -96,6 +102,7 @@ class LazyFilteredBTreeReaderThreadTest(unittest.TestCase):
index_path="/unused",
io_metas=io_metas,
executor=executor,
+ fallback_scan_max_size=256 * 1024 * 1024,
)
# Keep the patch active — reader_cls stays in module
return reader, executor, (mod, original)
@@ -106,6 +113,71 @@ class LazyFilteredBTreeReaderThreadTest(unittest.TestCase):
mod, original = patch_info
mod.BTreeIndexReader = original
+ def test_fallback_scan_budget_disables_range_scan(self):
+ """Range scans need fallback budget; point lookups remain direct."""
+ reader, executor, patch = self._create_reader(num_files=2, pool_size=2)
+ reader._fallback_scan_max_size = 0
+ field_ref = FieldRef(0, "id", "INT")
+ try:
+ self.assertIsNone(
+ reader.visit_less_than(field_ref, 15).result(timeout=5.0))
+ direct_result = reader.visit_equal(field_ref, 5).result(timeout=5.0)
+ self.assertIsNotNone(direct_result)
+ self.assertIn(5, list(direct_result.results()))
+ finally:
+ self._cleanup(reader, executor, patch)
+
+ def test_fallback_scan_budget_rejects_oversized_candidates(self):
+ reader, executor, patch = self._create_reader(num_files=2, pool_size=2)
+ reader._fallback_scan_max_size = 1024
+ field_ref = FieldRef(0, "id", "INT")
+ try:
+ self.assertIsNone(
+ reader.visit_less_than(field_ref, 15).result(timeout=5.0))
+ finally:
+ self._cleanup(reader, executor, patch)
+
+ def test_fallback_scan_budget_rejects_java_long_overflow(self):
+ reader, executor, patch = self._create_reader(num_files=2, pool_size=2)
+ reader._fallback_scan_max_size = (1 << 63) - 1
+ reader._selector._files = [
+ (_io_meta("large-0.index", 0, 9, (1 << 63) - 1),
+ BTreeIndexMeta.deserialize(_int_meta(0, 9))),
+ (_io_meta("large-1.index", 10, 19, 1),
+ BTreeIndexMeta.deserialize(_int_meta(10, 19))),
+ ]
+ field_ref = FieldRef(0, "id", "INT")
+ try:
+ self.assertIsNone(
+ reader.visit_less_than(field_ref, 15).result(timeout=5.0))
+ finally:
+ self._cleanup(reader, executor, patch)
+
+ def test_string_predicates_require_character_string_type(self):
+ reader, executor, patch = self._create_reader(num_files=1, pool_size=1)
+ try:
+ self.assertIsNone(
+ reader.visit_starts_with(FieldRef(0, "id", "BINARY"), b"a")
+ .result(timeout=5.0))
+ self.assertIsNone(
+ reader.visit_like(FieldRef(0, "id", "MAP<STRING, INT>"), "a%")
+ .result(timeout=5.0))
+ self.assertIsNotNone(
+ reader.visit_starts_with(FieldRef(0, "id", "VARCHAR(10) NOT NULL"), "1")
+ .result(timeout=5.0))
+ finally:
+ self._cleanup(reader, executor, patch)
+
+ def test_like_fallback_rejects_non_positive_budget_before_selection(self):
+ reader, executor, patch = self._create_reader(num_files=1, pool_size=1)
+ reader._fallback_scan_max_size = 0
+ try:
+ self.assertIsNone(
+ reader.visit_like(FieldRef(0, "id", "VARCHAR(10)"), "%1")
+ .result(timeout=5.0))
+ finally:
+ self._cleanup(reader, executor, patch)
+
def test_no_deadlock_more_files_than_threads(self):
"""With 8 files and only 2 threads, callback-based chaining must not deadlock."""
reader, executor, patch = self._create_reader(num_files=8, pool_size=2)
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 4f1df33585..5b2929cfac 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
@@ -25,6 +25,7 @@ import pyarrow as pa
from parameterized import parameterized
from pypaimon.catalog.catalog_factory import CatalogFactory
from pypaimon.data.generic_variant import GenericVariant
+from pypaimon.globalindex.global_index_scanner import GlobalIndexScanner
from pypaimon.schema.data_types import VectorType
from pypaimon.schema.schema import Schema
from pypaimon.read.read_builder import ReadBuilder
@@ -633,6 +634,148 @@ class JavaPyReadWriteTest(unittest.TestCase):
"partial append should not drop index manifest"
)
+ def test_read_bitmap_index_table(self):
+ table = self.catalog.get_table('default.test_bitmap_index_string')
+
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.equal('k', 'k2'),
+ [1, 2],
+ {'k': ['k2', 'k2'], 'v': ['v2', 'v2b']}
+ )
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.is_null('k'),
+ [4],
+ {'k': [None], 'v': ['v_null']},
+ schema=pa.schema([('k', pa.string()), ('v', pa.string())])
+ )
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.is_in('k', ['k1', 'k3']),
+ [0, 3],
+ {'k': ['k1', 'k3'], 'v': ['v1', 'v3']}
+ )
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.not_equal('k', 'k2'),
+ [0, 3],
+ {'k': ['k1', 'k3'], 'v': ['v1', 'v3']}
+ )
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.is_not_in('k', ['k1', 'k3']),
+ [1, 2],
+ {'k': ['k2', 'k2'], 'v': ['v2', 'v2b']}
+ )
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.between('k', 'k1', 'k2'),
+ [0, 1, 2],
+ {'k': ['k1', 'k2', 'k2'], 'v': ['v1', 'v2', 'v2b']}
+ )
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.not_between('k', 'k2', 'k2'),
+ [0, 3],
+ {'k': ['k1', 'k3'], 'v': ['v1', 'v3']}
+ )
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.startswith('k', 'k'),
+ [0, 1, 2, 3],
+ {'k': ['k1', 'k2', 'k2', 'k3'], 'v': ['v1', 'v2', 'v2b', 'v3']}
+ )
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.endswith('k', '2'),
+ [1, 2],
+ {'k': ['k2', 'k2'], 'v': ['v2', 'v2b']}
+ )
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.contains('k', '3'),
+ [3],
+ {'k': ['k3'], 'v': ['v3']}
+ )
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.like('k', 'k_'),
+ [0, 1, 2, 3],
+ {'k': ['k1', 'k2', 'k2', 'k3'], 'v': ['v1', 'v2', 'v2b', 'v3']}
+ )
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.equal('k', None),
+ [],
+ {'k': [], 'v': []},
+ schema=pa.schema([('k', pa.string()), ('v', pa.string())])
+ )
+ self._assert_bitmap_index_read(
+ table,
+ lambda builder: builder.is_not_in('k', ['k1', None]),
+ [],
+ {'k': [], 'v': []},
+ schema=pa.schema([('k', pa.string()), ('v', pa.string())])
+ )
+
+ def _assert_bitmap_index_read(self, table, predicate_factory, expected_row_ids,
+ expected_data, schema=None):
+ read_builder = table.new_read_builder()
+ predicate = predicate_factory(read_builder.new_predicate_builder())
+
+ scanner = GlobalIndexScanner.create(table, predicate=predicate)
+ self.assertIsNotNone(scanner)
+ with scanner:
+ result = scanner.scan(predicate)
+ self.assertIsNotNone(result)
+ self.assertEqual(expected_row_ids, sorted(list(result.results())))
+
+ scan = read_builder.new_scan().with_global_index_result(result)
+ actual = table_sort_by(read_builder.new_read().to_arrow(scan.plan().splits()), 'v')
+ expected = pa.Table.from_pydict(expected_data, schema=schema)
+ expected = table_sort_by(expected, 'v')
+ self.assertEqual(expected, actual)
+
+ def test_read_compressed_global_index_fallback_scan(self):
+ for compression in ('lz4', 'lzo'):
+ self._assert_compressed_global_index_fallback_scan(
+ 'test_btree_index_%s_fallback' % compression,
+ 'btree-index.fallback-scan-max-size')
+ self._assert_compressed_global_index_fallback_scan(
+ 'test_bitmap_index_%s_fallback' % compression,
+ 'bitmap-index.fallback-scan-max-size')
+
+ def _assert_compressed_global_index_fallback_scan(self, table_name, budget_key):
+ table = self.catalog.get_table('default.' + table_name)
+ predicate = (table.new_read_builder()
+ .new_predicate_builder()
+ .greater_or_equal('k', 'key-295'))
+
+ scanner = GlobalIndexScanner.create(table, predicate=predicate)
+ self.assertIsNotNone(scanner)
+ with scanner:
+ result = scanner.scan(predicate)
+ self.assertIsNotNone(result)
+ self.assertEqual([295, 296, 297, 298, 299],
+ sorted(list(result.results())))
+
+ read_builder = table.new_read_builder()
+ scan = read_builder.new_scan().with_global_index_result(result)
+ actual = table_sort_by(read_builder.new_read().to_arrow(scan.plan().splits()), 'k')
+ expected = pa.Table.from_pydict({
+ 'k': ['key-295', 'key-296', 'key-297', 'key-298', 'key-299'],
+ 'v': ['value-295', 'value-296', 'value-297', 'value-298', 'value-299'],
+ })
+ self.assertEqual(expected, actual)
+
+ disabled_table = table.copy({budget_key: '0 b'})
+ disabled_scanner = GlobalIndexScanner.create(disabled_table, predicate=predicate)
+ self.assertIsNotNone(disabled_scanner)
+ with disabled_scanner:
+ disabled_result = disabled_scanner.scan(predicate)
+ self.assertIsNone(disabled_result)
+
@parameterized.expand([('json',), ('csv',)])
def test_read_compressed_text_append_table(self, file_format):
table = self.catalog.get_table(
diff --git a/paimon-python/pypaimon/tests/sorted_index_file_meta_test.py b/paimon-python/pypaimon/tests/sorted_index_file_meta_test.py
new file mode 100644
index 0000000000..90d6ac9671
--- /dev/null
+++ b/paimon-python/pypaimon/tests/sorted_index_file_meta_test.py
@@ -0,0 +1,72 @@
+# 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 struct
+import unittest
+
+from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
+
+
+def _meta_bytes(first_key, last_key, has_nulls, flags=None):
+ first_bytes = first_key or b''
+ last_bytes = last_key or b''
+ data = (
+ struct.pack('<I', len(first_bytes)) + first_bytes +
+ struct.pack('<I', len(last_bytes)) + last_bytes +
+ struct.pack('<B', 1 if has_nulls else 0)
+ )
+ if flags is not None:
+ data += struct.pack('<B', SortedIndexFileMeta.FORMAT_VERSION_WITH_NULL_FLAGS)
+ data += struct.pack('<B', flags)
+ return data
+
+
+class SortedIndexFileMetaTest(unittest.TestCase):
+
+ def test_empty_string_key_is_not_null_with_format_flags(self):
+ meta = SortedIndexFileMeta.deserialize(_meta_bytes(b'', b'k9', False, 0))
+
+ self.assertEqual(b'', meta.first_key)
+ self.assertEqual(b'k9', meta.last_key)
+ self.assertFalse(meta.has_nulls)
+ self.assertFalse(meta.only_nulls())
+
+ def test_null_key_flags_are_honored(self):
+ meta = SortedIndexFileMeta.deserialize(
+ _meta_bytes(
+ None,
+ None,
+ True,
+ SortedIndexFileMeta.FIRST_KEY_IS_NULL
+ | SortedIndexFileMeta.LAST_KEY_IS_NULL))
+
+ self.assertIsNone(meta.first_key)
+ self.assertIsNone(meta.last_key)
+ self.assertTrue(meta.has_nulls)
+ self.assertTrue(meta.only_nulls())
+
+ def test_legacy_all_null_metadata(self):
+ meta = SortedIndexFileMeta.deserialize(_meta_bytes(None, None, True))
+
+ self.assertIsNone(meta.first_key)
+ self.assertIsNone(meta.last_key)
+ self.assertTrue(meta.has_nulls)
+ self.assertTrue(meta.only_nulls())
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py
index 5aa886220f..f2b453346e 100644
--- a/paimon-python/pypaimon/tests/vector_search_filter_test.py
+++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py
@@ -981,7 +981,7 @@ class VectorSearchFilterTest(unittest.TestCase):
class _FakeLazyReader:
def __init__(self_inner, key_serializer, file_io, index_path,
- io_metas, executor=None):
+ io_metas, executor=None, fallback_scan_max_size=None):
captured_io_metas.append(list(io_metas))
def close(self_inner):
@@ -1061,7 +1061,7 @@ class VectorSearchMultiShardScalarTest(unittest.TestCase):
"pypaimon.globalindex.btree.lazy_filtered_btree_reader.BTreeIndexReader",
_StubBTreeReader):
with mock.patch(
- "pypaimon.globalindex.btree.lazy_filtered_btree_reader.BTreeIndexMeta.deserialize",
+ "pypaimon.globalindex.sorted_file_global_index_reader.SortedIndexFileMeta.deserialize",
return_value=wide_meta):
scanner = GlobalIndexScanner(
fields=table.fields,
@@ -1183,8 +1183,8 @@ class VectorSearchMultiShardScalarTest(unittest.TestCase):
"pypaimon.globalindex.btree.lazy_filtered_btree_reader.BTreeIndexReader",
_StubBTreeReader):
with mock.patch(
- "pypaimon.globalindex.btree.lazy_filtered_btree_reader.BTreeIndexMeta.deserialize",
- return_value=BTreeIndexMeta(first_key=None, last_key=None, has_nulls=False)):
+ "pypaimon.globalindex.sorted_file_global_index_reader.SortedIndexFileMeta.deserialize",
+ return_value=BTreeIndexMeta(first_key=b'', last_key=b'zzzz', has_nulls=False)):
scanner = GlobalIndexScanner(
fields=table.fields,
file_io=table.file_io,
@@ -1194,11 +1194,11 @@ class VectorSearchMultiShardScalarTest(unittest.TestCase):
try:
result = scanner.scan(
Predicate(method="like", index=0, field="name",
- literals=["abc%"]))
+ literals=["a_c%"]))
finally:
scanner.close()
- self.assertEqual([("like", "abc%")], observed_calls)
+ self.assertEqual([("like", "a_c%")], observed_calls)
self.assertIsNotNone(result)
self.assertEqual([3], sorted(list(result.results())))