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 0c9811ad8c [python] Support bitmap global index build (#8402)
0c9811ad8c is described below
commit 0c9811ad8cde47f8a610224ca48f241202790bba
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 1 09:28:59 2026 +0800
[python] Support bitmap global index build (#8402)
Support building bitmap global indexes from PyPaimon, matching the Java
bitmap global index file format and sorted-index build path.
---
docs/docs/multimodal-table/global-index.mdx | 6 +-
docs/docs/multimodal-table/global-index/bitmap.mdx | 114 +++++++++
.../pypaimon/common/options/core_options.py | 36 +++
.../pypaimon/globalindex/bitmap/__init__.py | 6 +
.../globalindex/bitmap/bitmap_index_writer.py | 285 +++++++++++++++++++++
.../globalindex/btree/btree_index_writer.py | 74 ++----
.../pypaimon/globalindex/create_global_index.py | 61 +++--
.../pypaimon/globalindex/index_file_utils.py | 111 ++++++++
.../pypaimon/tests/global_index_build_test.py | 93 +++++++
9 files changed, 710 insertions(+), 76 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index.mdx
b/docs/docs/multimodal-table/global-index.mdx
index 6ee0882707..e80f5bce25 100644
--- a/docs/docs/multimodal-table/global-index.mdx
+++ b/docs/docs/multimodal-table/global-index.mdx
@@ -157,9 +157,9 @@ added_files = table.create_global_index(
</Tabs>
-PyPaimon global index build currently supports single-column BTree indexes and
-single-column paimon-vindex IVF vector indexes on tables with row tracking
-enabled.
+PyPaimon global index build currently supports single-column BTree indexes,
+single-column Bitmap indexes, and single-column paimon-vindex IVF vector
indexes
+on tables with row tracking enabled.
Drop index files:
diff --git a/docs/docs/multimodal-table/global-index/bitmap.mdx
b/docs/docs/multimodal-table/global-index/bitmap.mdx
index 453d8dbf00..154110af28 100644
--- a/docs/docs/multimodal-table/global-index/bitmap.mdx
+++ b/docs/docs/multimodal-table/global-index/bitmap.mdx
@@ -3,6 +3,9 @@ title: "Bitmap Index"
sidebar_position: 2
---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
@@ -56,6 +59,10 @@ exceeded, Paimon falls back to other matching indexes or
regular table scans.
## Build Bitmap Index
+<Tabs groupId="bitmap-build">
+
+<TabItem value="sql" label="SQL">
+
```sql
-- Create bitmap index on 'tag' column
CALL sys.create_global_index(
@@ -76,12 +83,90 @@ CALL sys.create_global_index(
);
```
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+table = catalog.get_table("db.my_table")
+
+# Create bitmap index on 'tag' column.
+added_files = table.create_global_index("tag", index_type="bitmap")
+print(added_files)
+```
+
+The API returns the number of committed index files. You can pass bitmap build
+options and restrict the build to selected partitions:
+
+```python
+added_files = table.create_global_index(
+ "tag",
+ index_type="bitmap",
+ partitions=[{"dt": "2026-06-18"}, {"dt": "2026-06-19"}],
+ options={
+ "sorted-index.records-per-range": "10000000",
+ "bitmap-index.dictionary-block-size": "16 kb",
+ },
+)
+print(added_files)
+```
+
+</TabItem>
+
+</Tabs>
+
Bitmap indexes share the sorted index build path with BTree indexes. Use
`sorted-index.records-per-range` to control the expected records per generated
index
file, and `sorted-index.build.max-parallelism` to cap Flink or Spark build
parallelism. The legacy `btree-index.records-per-range` and
`btree-index.build.max-parallelism` keys are still recognized as fallback keys.
+PyPaimon bitmap index build currently supports `bitmap-index.compression` set
to
+`none`, which is also the default. Use SQL if you need to build bitmap index
files
+with compressed dictionary blocks.
+
+## Drop Bitmap Index
+
+<Tabs groupId="bitmap-drop">
+
+<TabItem value="sql" label="SQL">
+
+```sql
+CALL sys.drop_global_index(
+ table => 'db.my_table',
+ index_column => 'tag',
+ index_type => 'bitmap'
+);
+```
+
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+table = catalog.get_table("db.my_table")
+
+dropped_files = table.drop_global_index("tag", index_type="bitmap")
+print(dropped_files)
+```
+
+You can also restrict the drop to selected partitions, or count matched files
+without committing:
+
+```python
+matched_files = table.drop_global_index(
+ "tag",
+ index_type="bitmap",
+ partitions=[{"dt": "2026-06-18"}, {"dt": "2026-06-19"}],
+ dry_run=True,
+)
+print(matched_files)
+```
+
+</TabItem>
+
+</Tabs>
+
## Bitmap Options
| Option | Default | Description |
@@ -98,10 +183,39 @@ parallelism. The legacy `btree-index.records-per-range` and
Once a bitmap index is built, it is automatically used during scan when a
filter
predicate matches the indexed column.
+<Tabs groupId="bitmap-search">
+
+<TabItem value="sql" label="SQL">
+
```sql
SELECT * FROM my_table WHERE tag IN ('vip', 'trial');
```
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+from pypaimon.common.predicate_builder import PredicateBuilder
+
+table = catalog.get_table("db.my_table")
+
+read_builder = table.new_read_builder()
+read_builder = read_builder.with_filter(
+ PredicateBuilder(table.fields)
+ .is_in("tag", ["vip", "trial"])
+)
+
+scan = read_builder.new_scan()
+read = read_builder.new_read()
+pa_table = read.to_arrow(scan.plan().splits())
+print(pa_table)
+```
+
+</TabItem>
+
+</Tabs>
+
For complement predicates such as `tag != 'blocked'` or
`tag NOT IN ('blocked', 'test')`, bitmap index evaluates the complement
against each
index file's own non-null row-id bitmap. This keeps results correct when one
logical
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index d76785198b..2dc8049601 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -725,6 +725,31 @@ class CoreOptions:
)
)
+ BITMAP_INDEX_DICTIONARY_BLOCK_SIZE: ConfigOption[MemorySize] = (
+ ConfigOptions.key("bitmap-index.dictionary-block-size")
+ .memory_type()
+ .default_value(MemorySize.of_kibi_bytes(16))
+ .with_description(
+ "The target dictionary block size for bitmap global indexes."
+ )
+ )
+
+ BITMAP_INDEX_COMPRESSION: ConfigOption[str] = (
+ ConfigOptions.key("bitmap-index.compression")
+ .string_type()
+ .default_value("none")
+ .with_description("Compression algorithm for bitmap global index
blocks.")
+ )
+
+ BITMAP_INDEX_COMPRESSION_LEVEL: ConfigOption[int] = (
+ ConfigOptions.key("bitmap-index.compression-level")
+ .int_type()
+ .default_value(1)
+ .with_description(
+ "Compression level for bitmap global index block compression."
+ )
+ )
+
LOCAL_CACHE_ENABLED: ConfigOption[bool] = (
ConfigOptions.key("local-cache.enabled")
.boolean_type()
@@ -1225,6 +1250,17 @@ class CoreOptions:
CoreOptions.BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE
).get_bytes()
+ def bitmap_index_dictionary_block_size(self) -> int:
+ return self.options.get(
+ CoreOptions.BITMAP_INDEX_DICTIONARY_BLOCK_SIZE
+ ).get_bytes()
+
+ def bitmap_index_compression(self) -> str:
+ return self.options.get(CoreOptions.BITMAP_INDEX_COMPRESSION)
+
+ def bitmap_index_compression_level(self) -> int:
+ return self.options.get(CoreOptions.BITMAP_INDEX_COMPRESSION_LEVEL)
+
def local_cache_enabled(self) -> bool:
return self.options.get(CoreOptions.LOCAL_CACHE_ENABLED)
diff --git a/paimon-python/pypaimon/globalindex/bitmap/__init__.py
b/paimon-python/pypaimon/globalindex/bitmap/__init__.py
index 4e37cce1eb..d3977a2c65 100644
--- a/paimon-python/pypaimon/globalindex/bitmap/__init__.py
+++ b/paimon-python/pypaimon/globalindex/bitmap/__init__.py
@@ -18,9 +18,15 @@
"""Bitmap global index support."""
from pypaimon.globalindex.bitmap.bitmap_index_reader import BitmapIndexReader
+from pypaimon.globalindex.bitmap.bitmap_index_writer import (
+ BITMAP_IDENTIFIER,
+ BitmapIndexWriter,
+)
from pypaimon.globalindex.bitmap.lazy_filtered_bitmap_reader import
LazyFilteredBitmapReader
__all__ = [
+ 'BITMAP_IDENTIFIER',
'BitmapIndexReader',
+ 'BitmapIndexWriter',
'LazyFilteredBitmapReader',
]
diff --git a/paimon-python/pypaimon/globalindex/bitmap/bitmap_index_writer.py
b/paimon-python/pypaimon/globalindex/bitmap/bitmap_index_writer.py
new file mode 100644
index 0000000000..c2cd24d7ec
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/bitmap/bitmap_index_writer.py
@@ -0,0 +1,285 @@
+# 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 writer compatible with Java's bitmap format."""
+
+from dataclasses import dataclass
+import struct
+from typing import Dict, List
+
+from pypaimon.globalindex.index_file_utils import (
+ BlockInfo,
+ PositionOutput,
+ new_global_index_file_name,
+ var_len_int_size,
+ var_len_long_size,
+ write_uncompressed_block,
+ write_var_len_int,
+ write_var_len_long,
+)
+from pypaimon.globalindex.result_entry import ResultEntry
+from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
+from pypaimon.utils.roaring_bitmap import RoaringBitmap64
+
+
+BITMAP_IDENTIFIER = "bitmap"
+_BITMAP_FOOTER_LENGTH = 48
+_BITMAP_MAGIC = 0x42474958
+_BITMAP_VERSION = 1
+_DEFAULT_DICTIONARY_BLOCK_SIZE = 16 * 1024
+
+
+@dataclass(frozen=True)
+class _DictionaryBlockMeta(BlockInfo):
+ first_key: bytes
+
+
+@dataclass(frozen=True)
+class _DictionaryEntry:
+ key: bytes
+ bitmap_block: BlockInfo
+
+ def estimated_size(self) -> int:
+ return (
+ var_len_int_size(len(self.key))
+ + len(self.key)
+ + var_len_long_size(self.bitmap_block.offset)
+ + var_len_int_size(self.bitmap_block.length)
+ )
+
+
+class _DictionaryBlockBuilder:
+ def __init__(self):
+ self.entries: List[_DictionaryEntry] = []
+ self._entries_size = 0
+
+ def has_entries(self) -> bool:
+ return len(self.entries) > 0
+
+ def estimated_size(self) -> int:
+ return var_len_int_size(len(self.entries)) + self._entries_size
+
+ def estimated_size_after(self, entry: _DictionaryEntry) -> int:
+ return (
+ var_len_int_size(len(self.entries) + 1)
+ + self._entries_size
+ + entry.estimated_size()
+ )
+
+ def add(self, entry: _DictionaryEntry) -> None:
+ self.entries.append(entry)
+ self._entries_size += entry.estimated_size()
+
+ def first_key(self) -> bytes:
+ return self.entries[0].key
+
+
+class BitmapIndexWriter:
+ """Writer for one bitmap global index file.
+
+ Row IDs are local to the manifest row range, matching the Java writer
+ contract used by sorted global index builders.
+ """
+
+ def __init__(
+ self,
+ file_io,
+ index_path: str,
+ key_serializer,
+ dictionary_block_size: int = _DEFAULT_DICTIONARY_BLOCK_SIZE,
+ compression: str = "none",
+ ):
+ if dictionary_block_size <= 0:
+ raise ValueError("bitmap-index.dictionary-block-size must be
positive.")
+ if compression is None:
+ compression = "none"
+ compression = str(compression).lower().strip()
+ if compression != "none":
+ raise ValueError(
+ "Python bitmap global index build currently supports only "
+ "bitmap-index.compression=none, got '%s'." % compression
+ )
+
+ self.file_name = new_global_index_file_name(BITMAP_IDENTIFIER)
+ self._file_io = file_io
+ self._index_path = index_path.rstrip("/")
+ self._key_serializer = key_serializer
+ self._comparator = key_serializer.create_comparator()
+ self._dictionary_block_size = dictionary_block_size
+ self._bitmaps: Dict[bytes, RoaringBitmap64] = {}
+ self._null_rows = RoaringBitmap64()
+ self._non_null_rows = RoaringBitmap64()
+ self._row_count = 0
+ self._first_key = None
+ self._last_key = None
+ self._closed = False
+
+ def write(self, key, row_id: int) -> None:
+ self._row_count += 1
+ if key is None:
+ self._null_rows.add(row_id)
+ return
+
+ self._non_null_rows.add(row_id)
+ self._update_min_max(key)
+ serialized_key = self._key_serializer.serialize(key)
+ bitmap = self._bitmaps.get(serialized_key)
+ if bitmap is None:
+ bitmap = RoaringBitmap64()
+ self._bitmaps[serialized_key] = bitmap
+ bitmap.add(row_id)
+
+ def finish(self) -> List[ResultEntry]:
+ if self._closed:
+ raise RuntimeError("BitmapIndexWriter is already closed.")
+ self._closed = True
+ if self._row_count == 0:
+ return []
+
+ self._file_io.check_or_mkdirs(self._index_path)
+ output = PositionOutput(
+ self._file_io.new_output_stream(self._file_path()))
+ try:
+ self._write(output)
+ output.close()
+ except Exception:
+ self._file_io.delete_quietly(self._file_path())
+ raise
+
+ meta = SortedIndexFileMeta(
+ None if self._first_key is None
+ else self._key_serializer.serialize(self._first_key),
+ None if self._last_key is None
+ else self._key_serializer.serialize(self._last_key),
+ not self._null_rows.is_empty(),
+ ).serialize()
+ return [ResultEntry(self.file_name, self._row_count, meta)]
+
+ def _write(self, output: PositionOutput) -> None:
+ null_rows_block = _write_bitmap_block(output, self._null_rows)
+ non_null_rows_block = _write_bitmap_block(output, self._non_null_rows)
+ dictionary_blocks, value_count =
self._write_dictionary_and_bitmap_blocks(
+ output)
+ index_block = _write_index_block(output, dictionary_blocks)
+ output.write(
+ _write_footer(
+ null_rows_block,
+ non_null_rows_block,
+ index_block,
+ value_count,
+ )
+ )
+
+ def _write_dictionary_and_bitmap_blocks(self, output: PositionOutput):
+ dictionary_blocks = []
+ current = _DictionaryBlockBuilder()
+ value_count = 0
+
+ for serialized_key, bitmap in sorted(self._bitmaps.items()):
+ bitmap_block = _write_bitmap_block(output, bitmap)
+ entry = _DictionaryEntry(serialized_key, bitmap_block)
+ if (
+ current.has_entries()
+ and current.estimated_size_after(entry) >
self._dictionary_block_size
+ ):
+ dictionary_blocks.append(_write_dictionary_block(output,
current))
+ current = _DictionaryBlockBuilder()
+ current.add(entry)
+ value_count += 1
+
+ if current.has_entries():
+ dictionary_blocks.append(_write_dictionary_block(output, current))
+ return dictionary_blocks, value_count
+
+ def _update_min_max(self, key) -> None:
+ if self._first_key is None or self._comparator(key, self._first_key) <
0:
+ self._first_key = key
+ if self._last_key is None or self._comparator(key, self._last_key) > 0:
+ self._last_key = key
+
+ def _file_path(self) -> str:
+ return "%s/%s" % (self._index_path, self.file_name)
+
+
+def _write_dictionary_block(
+ output: PositionOutput,
+ block: _DictionaryBlockBuilder,
+) -> _DictionaryBlockMeta:
+ data = bytearray()
+ data.extend(write_var_len_int(len(block.entries)))
+ for entry in block.entries:
+ data.extend(write_var_len_int(len(entry.key)))
+ data.extend(entry.key)
+ data.extend(write_var_len_long(entry.bitmap_block.offset))
+ data.extend(write_var_len_int(entry.bitmap_block.length))
+ block_info = _write_compressible_block(output, bytes(data))
+ return _DictionaryBlockMeta(
+ block_info.offset,
+ block_info.length,
+ block.first_key(),
+ )
+
+
+def _write_index_block(
+ output: PositionOutput,
+ blocks: List[_DictionaryBlockMeta],
+) -> BlockInfo:
+ data = bytearray()
+ data.extend(write_var_len_int(len(blocks)))
+ for block in blocks:
+ data.extend(write_var_len_int(len(block.first_key)))
+ data.extend(block.first_key)
+ data.extend(write_var_len_long(block.offset))
+ data.extend(write_var_len_int(block.length))
+ return _write_compressible_block(output, bytes(data))
+
+
+def _write_bitmap_block(
+ output: PositionOutput,
+ bitmap: RoaringBitmap64,
+) -> BlockInfo:
+ data = bitmap.serialize()
+ offset = output.pos
+ output.write(data)
+ return BlockInfo(offset, len(data))
+
+
+def _write_compressible_block(output: PositionOutput, data: bytes) ->
BlockInfo:
+ return write_uncompressed_block(output, data)
+
+
+def _write_footer(
+ null_rows_block: BlockInfo,
+ non_null_rows_block: BlockInfo,
+ index_block: BlockInfo,
+ value_count: int,
+) -> bytes:
+ result = struct.pack(
+ ">q i q i q i i i i",
+ null_rows_block.offset,
+ null_rows_block.length,
+ non_null_rows_block.offset,
+ non_null_rows_block.length,
+ index_block.offset,
+ index_block.length,
+ value_count,
+ _BITMAP_VERSION,
+ _BITMAP_MAGIC,
+ )
+ if len(result) != _BITMAP_FOOTER_LENGTH:
+ raise AssertionError("Unexpected bitmap footer length: %s" %
len(result))
+ return result
diff --git a/paimon-python/pypaimon/globalindex/btree/btree_index_writer.py
b/paimon-python/pypaimon/globalindex/btree/btree_index_writer.py
index 85be3fb2ee..be3dd7014a 100644
--- a/paimon-python/pypaimon/globalindex/btree/btree_index_writer.py
+++ b/paimon-python/pypaimon/globalindex/btree/btree_index_writer.py
@@ -18,13 +18,15 @@
"""BTree global index writer compatible with Java's SST-backed format."""
import struct
-import uuid
import zlib
from typing import List, Optional
-from pypaimon.globalindex.block_compression import (
- COMPRESSION_NONE,
- crc32c,
+from pypaimon.globalindex.index_file_utils import (
+ PositionOutput,
+ new_global_index_file_name,
+ write_uncompressed_block,
+ write_var_len_int,
+ write_var_len_long,
)
from pypaimon.globalindex.result_entry import ResultEntry
from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
@@ -39,47 +41,8 @@ _BLOCK_ALIGNED = 0
_BLOCK_UNALIGNED = 1
-def _write_var_len_int(value: int) -> bytes:
- if value < 0:
- raise ValueError("negative value: v=%s" % value)
- result = bytearray()
- while value & ~0x7F:
- result.append((value & 0x7F) | 0x80)
- value >>= 7
- result.append(value)
- return bytes(result)
-
-
-def _write_var_len_long(value: int) -> bytes:
- if value < 0:
- raise ValueError("negative value: v=%s" % value)
- result = bytearray()
- while value & ~0x7F:
- result.append((value & 0x7F) | 0x80)
- value >>= 7
- result.append(value)
- return bytes(result)
-
-
def _write_block_handle(offset: int, size: int) -> bytes:
- return _write_var_len_long(offset) + _write_var_len_int(size)
-
-
-class _PositionOutput:
- def __init__(self, output_stream):
- self._output_stream = output_stream
- self._pos = 0
-
- @property
- def pos(self) -> int:
- return self._pos
-
- def write(self, data: bytes) -> None:
- self._output_stream.write(data)
- self._pos += len(data)
-
- def close(self) -> None:
- self._output_stream.close()
+ return write_var_len_long(offset) + write_var_len_int(size)
class _BlockWriter:
@@ -97,9 +60,9 @@ class _BlockWriter:
def add(self, key: bytes, value: bytes) -> None:
start_position = len(self._block)
- self._block.extend(_write_var_len_int(len(key)))
+ self._block.extend(write_var_len_int(len(key)))
self._block.extend(key)
- self._block.extend(_write_var_len_int(len(value)))
+ self._block.extend(write_var_len_int(len(value)))
self._block.extend(value)
end_position = len(self._block)
@@ -137,7 +100,7 @@ class _BlockWriter:
class _SstFileWriter:
- def __init__(self, out: _PositionOutput, block_size: int):
+ def __init__(self, out: PositionOutput, block_size: int):
self._out = out
self._block_size = block_size
self._data_block_writer = _BlockWriter()
@@ -169,13 +132,9 @@ class _SstFileWriter:
def _write_block(self, block_writer: _BlockWriter):
block = block_writer.finish()
- block_offset = self._out.pos
- self._out.write(block)
- self._out.write(
- struct.pack('<B', COMPRESSION_NONE)
- + struct.pack('<I', crc32c(block, COMPRESSION_NONE)))
+ block_info = write_uncompressed_block(self._out, block)
block_writer.reset()
- return block_offset, len(block)
+ return block_info.offset, block_info.length
class BTreeIndexWriter:
@@ -192,8 +151,7 @@ class BTreeIndexWriter:
key_serializer,
block_size: int = 64 * 1024,
):
- self.file_name = (
- "%s-global-index-%s.index" % (BTREE_IDENTIFIER, uuid.uuid4()))
+ self.file_name = new_global_index_file_name(BTREE_IDENTIFIER)
self._file_io = file_io
self._index_path = index_path.rstrip('/')
self._key_serializer = key_serializer
@@ -206,7 +164,7 @@ class BTreeIndexWriter:
self._closed = False
self._file_io.check_or_mkdirs(self._index_path)
- self._output = _PositionOutput(
+ self._output = PositionOutput(
self._file_io.new_output_stream(self._file_path()))
self._sst = _SstFileWriter(self._output, block_size)
@@ -262,9 +220,9 @@ class BTreeIndexWriter:
return
value = bytearray()
- value.extend(_write_var_len_int(len(self._current_row_ids)))
+ value.extend(write_var_len_int(len(self._current_row_ids)))
for row_id in self._current_row_ids:
- value.extend(_write_var_len_long(row_id))
+ value.extend(write_var_len_long(row_id))
self._current_row_ids = []
self._sst.put(self._key_serializer.serialize(self._last_key),
bytes(value))
diff --git a/paimon-python/pypaimon/globalindex/create_global_index.py
b/paimon-python/pypaimon/globalindex/create_global_index.py
index b1c577005b..db184ddad9 100644
--- a/paimon-python/pypaimon/globalindex/create_global_index.py
+++ b/paimon-python/pypaimon/globalindex/create_global_index.py
@@ -29,6 +29,10 @@ from pypaimon.globalindex.btree.btree_index_writer import (
BTREE_IDENTIFIER,
BTreeIndexWriter,
)
+from pypaimon.globalindex.bitmap.bitmap_index_writer import (
+ BITMAP_IDENTIFIER,
+ BitmapIndexWriter,
+)
from pypaimon.globalindex.global_index_meta import GlobalIndexMeta
from pypaimon.globalindex.key_serializer import create_serializer
from pypaimon.globalindex.vindex.vindex_vector_global_index_reader import (
@@ -81,6 +85,10 @@ def create_global_index(
return sum(len(message.index_adds) for message in messages)
+_SORTED_INDEX_IDENTIFIERS = (BTREE_IDENTIFIER, BITMAP_IDENTIFIER)
+_SORTED_INDEX_RECORDS_PER_RANGE_FLOATING = 1.2
+
+
class GlobalIndexBuilder:
"""Small Python builder for global indexes."""
@@ -101,10 +109,13 @@ class GlobalIndexBuilder:
self._options = _merged_options(table, options)
self._core_options = CoreOptions(self._options)
- if self._index_type != BTREE_IDENTIFIER and self._index_type not in
VINDEX_IDENTIFIERS:
+ if (
+ self._index_type not in _SORTED_INDEX_IDENTIFIERS
+ and self._index_type not in VINDEX_IDENTIFIERS
+ ):
raise ValueError(
- "Python global index build currently supports '%s' and %s, got
'%s'."
- % (BTREE_IDENTIFIER, VINDEX_IDENTIFIERS, index_type)
+ "Python global index build currently supports %s and %s, got
'%s'."
+ % (_SORTED_INDEX_IDENTIFIERS, VINDEX_IDENTIFIERS, index_type)
)
if len(self._index_columns) != 1:
raise ValueError(
@@ -154,18 +165,23 @@ class GlobalIndexBuilder:
index_path_factory =
self._table.path_factory().global_index_path_factory()
index_path = index_path_factory.global_index_root_path()
- if self._index_type == BTREE_IDENTIFIER:
- return self._build_btree(splits, index_field, table_read,
index_path)
+ if self._index_type in _SORTED_INDEX_IDENTIFIERS:
+ return self._build_sorted_index(
+ splits, index_field, table_read, index_path)
return self._build_vindex(splits, index_field, table_read, index_path)
- def _build_btree(
+ def _build_sorted_index(
self, splits, index_field, table_read, index_path: str
) -> List[CommitMessage]:
key_serializer = create_serializer(index_field.type)
- block_size = self._core_options.btree_index_block_size()
- records_per_range = self._core_options.sorted_index_records_per_range()
- if records_per_range <= 0:
+ configured_records_per_range = (
+ self._core_options.sorted_index_records_per_range())
+ if configured_records_per_range <= 0:
raise ValueError("sorted-index.records-per-range must be
positive.")
+ records_per_range = int(
+ configured_records_per_range
+ * _SORTED_INDEX_RECORDS_PER_RANGE_FLOATING
+ )
messages = []
for split in _split_by_contiguous_row_range(splits):
@@ -183,12 +199,8 @@ class GlobalIndexBuilder:
continue
index_adds = []
for chunk in _chunks(rows, records_per_range):
- writer = BTreeIndexWriter(
- self._table.file_io,
- index_path,
- key_serializer,
- block_size=block_size,
- )
+ writer = self._create_sorted_index_writer(
+ index_path, key_serializer)
for key, row_id in chunk:
writer.write(key, row_id - row_range.from_)
result_entries = writer.finish()
@@ -213,6 +225,25 @@ class GlobalIndexBuilder:
)
return messages
+ def _create_sorted_index_writer(self, index_path: str, key_serializer):
+ if self._index_type == BTREE_IDENTIFIER:
+ return BTreeIndexWriter(
+ self._table.file_io,
+ index_path,
+ key_serializer,
+ block_size=self._core_options.btree_index_block_size(),
+ )
+ if self._index_type == BITMAP_IDENTIFIER:
+ return BitmapIndexWriter(
+ self._table.file_io,
+ index_path,
+ key_serializer,
+ dictionary_block_size=(
+ self._core_options.bitmap_index_dictionary_block_size()),
+ compression=self._core_options.bitmap_index_compression(),
+ )
+ raise ValueError("Unsupported sorted global index type: %s" %
self._index_type)
+
def _build_vindex(
self, splits, index_field, table_read, index_path: str
) -> List[CommitMessage]:
diff --git a/paimon-python/pypaimon/globalindex/index_file_utils.py
b/paimon-python/pypaimon/globalindex/index_file_utils.py
new file mode 100644
index 0000000000..a27a923b04
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/index_file_utils.py
@@ -0,0 +1,111 @@
+# 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 helpers for Java-compatible global index files."""
+
+from dataclasses import dataclass
+import struct
+import uuid
+
+from pypaimon.globalindex.block_compression import COMPRESSION_NONE, crc32c
+
+
+@dataclass(frozen=True)
+class BlockInfo:
+ offset: int
+ length: int
+
+
+class PositionOutput:
+ def __init__(self, output_stream):
+ self._output_stream = output_stream
+ self._pos = 0
+
+ @property
+ def pos(self) -> int:
+ return self._pos
+
+ def write(self, data: bytes) -> None:
+ self._output_stream.write(data)
+ self._pos += len(data)
+
+ def close(self) -> None:
+ self._output_stream.close()
+
+
+def new_global_index_file_name(prefix: str) -> str:
+ return "%s-global-index-%s.index" % (prefix, uuid.uuid4())
+
+
+def write_uncompressed_block(output: PositionOutput, data: bytes) -> BlockInfo:
+ offset = output.pos
+ output.write(data)
+ write_block_trailer(output, data, COMPRESSION_NONE)
+ return BlockInfo(offset, len(data))
+
+
+def write_block_trailer(
+ output: PositionOutput,
+ data: bytes,
+ compression_type: int,
+) -> None:
+ output.write(
+ struct.pack("<B", compression_type)
+ + struct.pack("<I", crc32c(data, compression_type))
+ )
+
+
+def write_var_len_int(value: int) -> bytes:
+ if value < 0:
+ raise ValueError("negative value: v=%s" % value)
+ result = bytearray()
+ while value & ~0x7F:
+ result.append((value & 0x7F) | 0x80)
+ value >>= 7
+ result.append(value)
+ return bytes(result)
+
+
+def write_var_len_long(value: int) -> bytes:
+ if value < 0:
+ raise ValueError("negative value: v=%s" % value)
+ result = bytearray()
+ while value & ~0x7F:
+ result.append((value & 0x7F) | 0x80)
+ value >>= 7
+ result.append(value)
+ return bytes(result)
+
+
+def var_len_int_size(value: int) -> int:
+ if value < 0:
+ raise ValueError("negative value: v=%s" % value)
+ size = 1
+ while value & ~0x7F:
+ value >>= 7
+ size += 1
+ return size
+
+
+def var_len_long_size(value: int) -> int:
+ if value < 0:
+ raise ValueError("negative value: v=%s" % value)
+ size = 1
+ while value & ~0x7F:
+ value >>= 7
+ size += 1
+ return size
diff --git a/paimon-python/pypaimon/tests/global_index_build_test.py
b/paimon-python/pypaimon/tests/global_index_build_test.py
index 00279d01d7..957f6583be 100644
--- a/paimon-python/pypaimon/tests/global_index_build_test.py
+++ b/paimon-python/pypaimon/tests/global_index_build_test.py
@@ -221,6 +221,99 @@ class GlobalIndexBuildTest(
index_read_builder.new_scan().plan().splits())
self.assertEqual(0, index_table.num_rows)
+ def test_create_bitmap_global_index_from_python(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.table(
+ {
+ 'id': [1, 2, 3, 4, 5],
+ 'name': ['a', 'b', 'c', 'd', 'e'],
+ 'age': [10, 20, 30, 40, 50],
+ 'city': ['vip', 'trial', None, 'vip', 'blocked'],
+ },
+ schema=self.pa_schema,
+ ))
+
+ added = table.create_global_index(
+ 'city',
+ index_type='bitmap',
+ options={
+ 'sorted-index.records-per-range': '2',
+ 'bitmap-index.dictionary-block-size': '1 b',
+ },
+ )
+
+ self.assertEqual(3, added)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ entries = IndexFileHandler(table).scan(snapshot)
+ self.assertEqual(3, len(entries))
+ self.assertEqual({'bitmap'}, {e.index_file.index_type for e in
entries})
+ self.assertEqual([1, 2, 2],
+ sorted(e.index_file.row_count for e in entries))
+ self.assertEqual(
+ {0},
+ {e.index_file.global_index_meta.row_range_start for e in entries},
+ )
+ self.assertEqual(
+ {4},
+ {e.index_file.global_index_meta.row_range_end for e in entries},
+ )
+
+ read_builder = table.new_read_builder()
+ predicate_builder = read_builder.new_predicate_builder()
+ cases = [
+ (predicate_builder.is_in('city', ['vip', 'trial']),
+ [Range(0, 1), Range(3, 3)]),
+ (predicate_builder.is_null('city'), [Range(2, 2)]),
+ (predicate_builder.not_equal('city', 'blocked'),
+ [Range(0, 1), Range(3, 3)]),
+ ]
+ for predicate, expected in cases:
+ with GlobalIndexScanner.create(
+ table,
+ predicate=predicate,
+ snapshot=snapshot) as scanner:
+ result = scanner.scan(predicate)
+ self.assertEqual(expected, result.results().to_range_list())
+
+ def test_create_bitmap_global_index_rejects_unsupported_compression(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.table(
+ {
+ 'id': [1],
+ 'name': ['a'],
+ 'age': [10],
+ 'city': ['vip'],
+ },
+ schema=self.pa_schema,
+ ))
+
+ with self.assertRaisesRegex(ValueError,
'bitmap-index.compression=none'):
+ table.create_global_index(
+ 'city',
+ index_type='bitmap',
+ options={'bitmap-index.compression': 'lz4'},
+ )
+
+ def test_sorted_index_records_per_range_matches_java_floating_factor(self):
+ table = self._create_table()
+ rows = list(range(12))
+ self._write_arrow(table, pa.table(
+ {
+ 'id': rows,
+ 'name': ['n%s' % i for i in rows],
+ 'age': rows,
+ 'city': ['c%s' % i for i in rows],
+ },
+ schema=self.pa_schema,
+ ))
+
+ added = table.create_global_index(
+ 'id',
+ options={'sorted-index.records-per-range': '10'},
+ )
+
+ self.assertEqual(1, added)
+
def test_create_global_index_uses_external_path(self):
external_root = 'file://%s' % os.path.join(
self.tempdir, 'global-index-external')
