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 0abb9c3c0c [python] Add global index lifecycle APIs (#8393)
0abb9c3c0c is described below
commit 0abb9c3c0c42af527b107506d1f496a8ef66cfb9
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jun 30 22:03:55 2026 +0800
[python] Add global index lifecycle APIs (#8393)
Add PyPaimon support for building and dropping BTree global indexes, and
expose the metadata needed to inspect global index coverage from Python.
---
docs/docs/multimodal-table/global-index.mdx | 239 ++++++++++++-
.../pypaimon/common/options/core_options.py | 53 +++
paimon-python/pypaimon/globalindex/__init__.py | 12 +
.../pypaimon/globalindex/btree/__init__.py | 2 +
.../globalindex/btree/btree_index_writer.py | 310 +++++++++++++++++
.../pypaimon/globalindex/create_global_index.py | 386 +++++++++++++++++++++
.../pypaimon/globalindex/drop_global_index.py | 177 ++++++++++
.../pypaimon/globalindex/key_serializer.py | 323 ++++++++++++++++-
.../{btree/__init__.py => result_entry.py} | 22 +-
.../pypaimon/globalindex/sorted_index_file_meta.py | 23 ++
.../pypaimon/manifest/index_manifest_file.py | 81 ++++-
paimon-python/pypaimon/table/file_store_table.py | 27 ++
.../pypaimon/table/system/file_key_ranges_table.py | 145 ++++++++
.../pypaimon/table/system/system_table_loader.py | 9 +-
.../pypaimon/table/system/table_indexes_table.py | 169 +++++++++
.../pypaimon/tests/global_index_build_test.py | 292 ++++++++++++++++
.../tests/system/system_table_loader_test.py | 4 +-
.../pypaimon/utils/file_store_path_factory.py | 32 +-
paimon-python/pypaimon/write/commit_message.py | 8 +-
paimon-python/pypaimon/write/file_store_commit.py | 33 +-
20 files changed, 2304 insertions(+), 43 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index.mdx
b/docs/docs/multimodal-table/global-index.mdx
index 1cecce7993..a163390e0b 100644
--- a/docs/docs/multimodal-table/global-index.mdx
+++ b/docs/docs/multimodal-table/global-index.mdx
@@ -3,6 +3,9 @@ title: "Global Index"
sidebar_position: 8
---
+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
@@ -55,6 +58,10 @@ Global indexes work on top of Data Evolution tables. To use
global indexes, your
Create a table with the required properties:
+<Tabs groupId="global-index-create-table">
+
+<TabItem value="sql" label="SQL">
+
```sql
CREATE TABLE my_table (
id INT,
@@ -69,10 +76,44 @@ CREATE TABLE my_table (
);
```
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+import pyarrow as pa
+
+from pypaimon import Schema
+
+schema = Schema.from_pyarrow_schema(
+ pa.schema([
+ pa.field("id", pa.int32()),
+ pa.field("name", pa.string()),
+ pa.field("embedding", pa.list_(pa.float32())),
+ pa.field("content", pa.string()),
+ ]),
+ options={
+ "bucket": "-1",
+ "row-tracking.enabled": "true",
+ "data-evolution.enabled": "true",
+ "global-index.enabled": "true",
+ },
+)
+
+catalog.create_table("db.my_table", schema, ignore_if_exists=False)
+```
+
+</TabItem>
+
+</Tabs>
+
## Lifecycle
-Create global indexes with the `create_global_index` procedure. You can build
an index for all
-partitions or only selected partitions:
+Create global indexes for all partitions or only selected partitions:
+
+<Tabs groupId="global-index-build">
+
+<TabItem value="sql" label="SQL">
```sql
CALL sys.create_global_index(
@@ -89,7 +130,41 @@ CALL sys.create_global_index(
);
```
-Drop index files with `drop_global_index`:
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+table = catalog.get_table("db.my_table")
+
+added_files = table.create_global_index("name")
+print(added_files)
+```
+
+The API returns the number of committed index files. You can pass build
options and restrict the
+build to selected partitions:
+
+```python
+added_files = table.create_global_index(
+ "name",
+ index_type="btree",
+ partitions=[{"dt": "2026-06-18"}, {"dt": "2026-06-19"}],
+ options={"sorted-index.records-per-range": "10000000"},
+)
+```
+
+</TabItem>
+
+</Tabs>
+
+PyPaimon global index build currently supports single-column BTree indexes on
+tables with row tracking enabled.
+
+Drop index files:
+
+<Tabs groupId="global-index-drop">
+
+<TabItem value="sql" label="SQL">
```sql
CALL sys.drop_global_index(
@@ -99,25 +174,111 @@ CALL sys.drop_global_index(
);
```
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+table = catalog.get_table("db.my_table")
+
+dropped_files = table.drop_global_index("name", index_type="btree")
+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(
+ "name",
+ index_type="btree",
+ partitions=[{"dt": "2026-06-18"}, {"dt": "2026-06-19"}],
+ dry_run=True,
+)
+```
+
+</TabItem>
+
+</Tabs>
+
Global indexes are stored in index files and recorded in table metadata. To
inspect index files and
their row-id coverage, query the `table_indexes` system table:
+<Tabs groupId="global-index-table-indexes">
+
+<TabItem value="sql" label="SQL">
+
```sql
SELECT index_type, index_field_name, row_range_start, row_range_end
FROM my_table$table_indexes
WHERE index_field_name IS NOT NULL;
```
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+import pyarrow.compute as pc
+
+table_indexes = catalog.get_table("db.my_table$table_indexes")
+read_builder = table_indexes.new_read_builder().with_projection([
+ "index_type",
+ "index_field_name",
+ "row_range_start",
+ "row_range_end",
+])
+
+pa_table = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits()
+)
+pa_table = pa_table.filter(pc.is_valid(pa_table["index_field_name"]))
+print(pa_table)
+```
+
+</TabItem>
+
+</Tabs>
+
You can also query `file_key_ranges` to inspect data file row-id ranges and
diagnose coverage:
+<Tabs groupId="global-index-file-key-ranges">
+
+<TabItem value="sql" label="SQL">
+
```sql
SELECT file_path, first_row_id, record_count
FROM my_table$file_key_ranges;
```
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+file_key_ranges = catalog.get_table("db.my_table$file_key_ranges")
+read_builder = file_key_ranges.new_read_builder().with_projection([
+ "file_path",
+ "first_row_id",
+ "record_count",
+])
+
+pa_table = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits()
+)
+print(pa_table)
+```
+
+</TabItem>
+
+</Tabs>
+
For workloads that need newly appended rows to become visible only after
existing global indexes
cover them, enable the visibility callback:
+<Tabs groupId="global-index-visibility-callback">
+
+<TabItem value="sql" label="SQL">
+
```sql
ALTER TABLE my_table SET (
'visibility-callback.enabled' = 'true',
@@ -125,19 +286,66 @@ ALTER TABLE my_table SET (
);
```
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+from pypaimon.schema.schema_change import SchemaChange
+
+catalog.alter_table(
+ "db.my_table",
+ [
+ SchemaChange.set_option("visibility-callback.enabled", "true"),
+ SchemaChange.set_option("visibility-callback.timeout", "30 min"),
+ ],
+)
+```
+
+</TabItem>
+
+</Tabs>
+
## Coverage and Freshness
Global index files cover row-id ranges. If more rows are appended after an
index is built, those
-new rows are not automatically covered by the existing index files. Run
`create_global_index` again
-to build index files for newly uncovered data. By default, queries use fast
search and only read
+new rows are not automatically covered by the existing index files. Build the
global index again
+to create index files for newly uncovered data. By default, queries use fast
search and only read
indexed row ranges; rows in uncovered ranges are not returned for that indexed
query.
To improve freshness for query types that support raw-data search, set:
+<Tabs groupId="global-index-search-mode">
+
+<TabItem value="sql" label="SQL">
+
```sql
ALTER TABLE my_table SET ('global-index.search-mode' = 'full');
```
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+from pypaimon.schema.schema_change import SchemaChange
+
+catalog.alter_table(
+ "db.my_table",
+ [SchemaChange.set_option("global-index.search-mode", "full")],
+)
+```
+
+For a read-only override on an existing `Table` instance:
+
+```python
+full_table = table.copy({"global-index.search-mode": "full"})
+```
+
+</TabItem>
+
+</Tabs>
+
With `full` search, supported global-index queries first use the snapshot
`nextRowId` and global
index row-id coverage to detect whether any row range is missing from the
index. Raw data is scanned
only when such a gap exists. Use `detail` search when data files may have been
rewritten or updated
@@ -146,10 +354,31 @@ handle index invalidation caused by updates or rewrites.
To temporarily disable global-index scan acceleration while keeping the index
files, set:
+<Tabs groupId="global-index-enabled">
+
+<TabItem value="sql" label="SQL">
+
```sql
ALTER TABLE my_table SET ('global-index.enabled' = 'false');
```
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+from pypaimon.schema.schema_change import SchemaChange
+
+catalog.alter_table(
+ "db.my_table",
+ [SchemaChange.set_option("global-index.enabled", "false")],
+)
+```
+
+</TabItem>
+
+</Tabs>
+
Set it back to `true` to use global indexes during scans again.
## Build Options
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 56e323c997..9ae32c187b 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -20,6 +20,7 @@ import warnings
from datetime import timedelta
from enum import Enum
from typing import Dict, List, Optional
+from urllib.parse import urlparse
from pypaimon.common.memory_size import MemorySize
from pypaimon.common.options import Options
@@ -646,6 +647,16 @@ class CoreOptions:
)
)
+ GLOBAL_INDEX_EXTERNAL_PATH: ConfigOption[str] = (
+ ConfigOptions.key("global-index.external-path")
+ .string_type()
+ .no_default_value()
+ .with_description(
+ "Global index root directory. If not set, global index files are "
+ "stored under the table index directory."
+ )
+ )
+
GLOBAL_INDEX_THREAD_NUM: ConfigOption[int] = (
ConfigOptions.key("global-index.thread-num")
.int_type()
@@ -675,6 +686,29 @@ class CoreOptions:
)
)
+ BTREE_INDEX_BLOCK_SIZE: ConfigOption[MemorySize] = (
+ ConfigOptions.key("btree-index.block-size")
+ .memory_type()
+ .default_value(MemorySize.of_kibi_bytes(64))
+ .with_description("The block size to use for BTree global indexes.")
+ )
+
+ SORTED_INDEX_RECORDS_PER_RANGE: ConfigOption[int] = (
+ ConfigOptions.key("sorted-index.records-per-range")
+ .long_type()
+ .default_value(10_000_000)
+ .with_description("The expected number of records per sorted global
index file.")
+ )
+
+ BTREE_INDEX_RECORDS_PER_RANGE: ConfigOption[int] = (
+ ConfigOptions.key("btree-index.records-per-range")
+ .long_type()
+ .default_value(10_000_000)
+ .with_description(
+ "The expected number of records per BTree global index file."
+ )
+ )
+
BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE: ConfigOption[MemorySize] = (
ConfigOptions.key("bitmap-index.fallback-scan-max-size")
.memory_type()
@@ -1149,6 +1183,17 @@ class CoreOptions:
def global_index_search_mode(self):
return self.options.get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE)
+ def global_index_external_path(self, default=None):
+ value = self.options.get(CoreOptions.GLOBAL_INDEX_EXTERNAL_PATH,
default)
+ if value is None:
+ return None
+ value = str(value).strip()
+ if not value:
+ return None
+ if not urlparse(value).scheme:
+ raise ValueError("scheme should not be null: %s" % value)
+ return value
+
def global_index_thread_num(self) -> Optional[int]:
return self.options.get(CoreOptions.GLOBAL_INDEX_THREAD_NUM)
@@ -1157,6 +1202,14 @@ class CoreOptions:
CoreOptions.BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE
).get_bytes()
+ def btree_index_block_size(self) -> int:
+ return self.options.get(CoreOptions.BTREE_INDEX_BLOCK_SIZE).get_bytes()
+
+ def sorted_index_records_per_range(self) -> int:
+ if self.options.contains(CoreOptions.SORTED_INDEX_RECORDS_PER_RANGE):
+ return self.options.get(CoreOptions.SORTED_INDEX_RECORDS_PER_RANGE)
+ return self.options.get(CoreOptions.BTREE_INDEX_RECORDS_PER_RANGE)
+
def bitmap_index_fallback_scan_max_size(self) -> int:
return self.options.get(
CoreOptions.BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE
diff --git a/paimon-python/pypaimon/globalindex/__init__.py
b/paimon-python/pypaimon/globalindex/__init__.py
index 35c043f577..54b8c65f86 100644
--- a/paimon-python/pypaimon/globalindex/__init__.py
+++ b/paimon-python/pypaimon/globalindex/__init__.py
@@ -44,6 +44,14 @@ from pypaimon.globalindex.offset_global_index_reader import
OffsetGlobalIndexRea
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.globalindex.create_global_index import (
+ GlobalIndexBuilder,
+ create_global_index,
+)
+from pypaimon.globalindex.drop_global_index import (
+ GlobalIndexDropper,
+ drop_global_index,
+)
from pypaimon.utils.range import Range
__all__ = [
@@ -74,5 +82,9 @@ __all__ = [
'SortedFileGlobalIndexReader',
'SortedFileMetaSelector',
'SortedIndexFileMeta',
+ 'GlobalIndexBuilder',
+ 'create_global_index',
+ 'GlobalIndexDropper',
+ 'drop_global_index',
'Range',
]
diff --git a/paimon-python/pypaimon/globalindex/btree/__init__.py
b/paimon-python/pypaimon/globalindex/btree/__init__.py
index 17a6a30cbc..62720f380d 100644
--- a/paimon-python/pypaimon/globalindex/btree/__init__.py
+++ b/paimon-python/pypaimon/globalindex/btree/__init__.py
@@ -20,11 +20,13 @@
from pypaimon.globalindex.btree.btree_index_reader import BTreeIndexReader
from pypaimon.globalindex.btree.btree_index_meta import BTreeIndexMeta
from pypaimon.globalindex.btree.btree_file_meta_selector import
BTreeFileMetaSelector
+from pypaimon.globalindex.btree.btree_index_writer import BTreeIndexWriter
from pypaimon.globalindex.btree.key_serializer import KeySerializer
from pypaimon.globalindex.btree.lazy_filtered_btree_reader import
LazyFilteredBTreeReader
__all__ = [
'BTreeIndexReader',
+ 'BTreeIndexWriter',
'BTreeIndexMeta',
'BTreeFileMetaSelector',
'KeySerializer',
diff --git a/paimon-python/pypaimon/globalindex/btree/btree_index_writer.py
b/paimon-python/pypaimon/globalindex/btree/btree_index_writer.py
new file mode 100644
index 0000000000..85be3fb2ee
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/btree/btree_index_writer.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.
+
+"""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.result_entry import ResultEntry
+from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
+from pypaimon.utils.roaring_bitmap import RoaringBitmap64
+
+
+BTREE_IDENTIFIER = "btree"
+_BTREE_FOOTER_LENGTH = 52
+_BTREE_MAGIC_NUMBER = 0x50425449
+_BTREE_CURRENT_VERSION = 1
+_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()
+
+
+class _BlockWriter:
+ def __init__(self):
+ self._positions: List[int] = []
+ self._block = bytearray()
+ self._aligned_size = 0
+ self._aligned = True
+
+ def reset(self) -> None:
+ self._positions = []
+ self._block = bytearray()
+ self._aligned_size = 0
+ self._aligned = True
+
+ 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(key)
+ self._block.extend(_write_var_len_int(len(value)))
+ self._block.extend(value)
+ end_position = len(self._block)
+
+ self._positions.append(start_position)
+ if self._aligned:
+ current_size = end_position - start_position
+ if self._aligned_size == 0:
+ self._aligned_size = current_size
+ else:
+ self._aligned = self._aligned_size == current_size
+
+ def size(self) -> int:
+ return len(self._positions)
+
+ def memory(self) -> int:
+ size = len(self._block) + 5
+ if not self._aligned:
+ size += len(self._positions) * 4
+ return size
+
+ def finish(self) -> bytes:
+ if not self._positions:
+ self._aligned = False
+
+ result = bytearray(self._block)
+ if self._aligned:
+ result.extend(struct.pack('<I', self._aligned_size))
+ result.append(_BLOCK_ALIGNED)
+ else:
+ for position in self._positions:
+ result.extend(struct.pack('<I', position))
+ result.extend(struct.pack('<I', len(self._positions)))
+ result.append(_BLOCK_UNALIGNED)
+ return bytes(result)
+
+
+class _SstFileWriter:
+ def __init__(self, out: _PositionOutput, block_size: int):
+ self._out = out
+ self._block_size = block_size
+ self._data_block_writer = _BlockWriter()
+ self._index_block_writer = _BlockWriter()
+ self._last_key: Optional[bytes] = None
+
+ def put(self, key: bytes, value: bytes) -> None:
+ self._data_block_writer.add(key, value)
+ self._last_key = key
+
+ if self._data_block_writer.memory() > self._block_size:
+ self.flush()
+
+ def flush(self) -> None:
+ if self._data_block_writer.size() == 0:
+ return
+ block_offset, block_size = self._write_block(self._data_block_writer)
+ self._index_block_writer.add(
+ self._last_key, _write_block_handle(block_offset, block_size))
+
+ def write_bloom_filter(self):
+ return None
+
+ def write_index_block(self):
+ return self._write_block(self._index_block_writer)
+
+ def write_slice(self, data: bytes) -> None:
+ self._out.write(data)
+
+ 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_writer.reset()
+ return block_offset, len(block)
+
+
+class BTreeIndexWriter:
+ """Writer for one BTree global index file.
+
+ Keys must be written in non-decreasing order, matching Java's writer
+ contract. Row IDs are local to the manifest row range.
+ """
+
+ def __init__(
+ self,
+ file_io,
+ index_path: str,
+ key_serializer,
+ block_size: int = 64 * 1024,
+ ):
+ self.file_name = (
+ "%s-global-index-%s.index" % (BTREE_IDENTIFIER, uuid.uuid4()))
+ self._file_io = file_io
+ self._index_path = index_path.rstrip('/')
+ self._key_serializer = key_serializer
+ self._comparator = key_serializer.create_comparator()
+ self._current_row_ids: List[int] = []
+ self._null_bitmap: Optional[RoaringBitmap64] = None
+ self._first_key = None
+ self._last_key = None
+ self._row_count = 0
+ self._closed = False
+
+ self._file_io.check_or_mkdirs(self._index_path)
+ self._output = _PositionOutput(
+ self._file_io.new_output_stream(self._file_path()))
+ self._sst = _SstFileWriter(self._output, block_size)
+
+ def write(self, key, row_id: int) -> None:
+ self._row_count += 1
+ if key is None:
+ if self._null_bitmap is None:
+ self._null_bitmap = RoaringBitmap64()
+ self._null_bitmap.add(row_id)
+ return
+
+ if self._last_key is not None and self._comparator(key,
self._last_key) != 0:
+ self._flush()
+
+ self._last_key = key
+ self._current_row_ids.append(row_id)
+ if self._first_key is None:
+ self._first_key = key
+
+ def finish(self) -> List[ResultEntry]:
+ if self._closed:
+ raise RuntimeError("BTreeIndexWriter is already closed.")
+ self._closed = True
+
+ try:
+ self._flush()
+ self._sst.flush()
+ null_bitmap_handle = self._write_null_bitmap()
+ bloom_filter_handle = self._sst.write_bloom_filter()
+ index_block_handle = self._sst.write_index_block()
+ self._sst.write_slice(
+ _write_footer(
+ bloom_filter_handle, index_block_handle,
null_bitmap_handle))
+ self._output.close()
+ except Exception:
+ self._file_io.delete_quietly(self._file_path())
+ raise
+
+ if self._first_key is None and self._null_bitmap is None:
+ raise RuntimeError("Should never write an empty btree index file.")
+
+ 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),
+ self._null_bitmap is not None,
+ ).serialize()
+ return [ResultEntry(self.file_name, self._row_count, meta)]
+
+ def _flush(self) -> None:
+ if not self._current_row_ids:
+ return
+
+ value = bytearray()
+ 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))
+ self._current_row_ids = []
+ self._sst.put(self._key_serializer.serialize(self._last_key),
bytes(value))
+
+ def _write_null_bitmap(self):
+ if self._null_bitmap is None:
+ return None
+
+ serialized = self._null_bitmap.serialize()
+ offset = self._output.pos
+ self._output.write(serialized)
+ self._output.write(struct.pack('<I', zlib.crc32(serialized) &
0xFFFFFFFF))
+ return offset, len(serialized)
+
+ def _file_path(self) -> str:
+ return "%s/%s" % (self._index_path, self.file_name)
+
+
+def _write_footer(bloom_filter_handle, index_block_handle, null_bitmap_handle)
-> bytes:
+ result = bytearray()
+ if bloom_filter_handle is None:
+ result.extend(struct.pack('<QIQ', 0, 0, 0))
+ else:
+ result.extend(
+ struct.pack(
+ '<QIQ',
+ bloom_filter_handle[0],
+ bloom_filter_handle[1],
+ bloom_filter_handle[2],
+ ))
+
+ result.extend(struct.pack('<QI', index_block_handle[0],
index_block_handle[1]))
+
+ if null_bitmap_handle is None:
+ result.extend(struct.pack('<QI', 0, 0))
+ else:
+ result.extend(
+ struct.pack('<QI', null_bitmap_handle[0], null_bitmap_handle[1]))
+
+ result.extend(struct.pack('<I', _BTREE_CURRENT_VERSION))
+ result.extend(struct.pack('<I', _BTREE_MAGIC_NUMBER))
+ if len(result) != _BTREE_FOOTER_LENGTH:
+ raise AssertionError("Unexpected BTree footer length: %s" %
len(result))
+ return bytes(result)
diff --git a/paimon-python/pypaimon/globalindex/create_global_index.py
b/paimon-python/pypaimon/globalindex/create_global_index.py
new file mode 100644
index 0000000000..3aa3bb919f
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/create_global_index.py
@@ -0,0 +1,386 @@
+# 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.
+
+"""Build global index files from Python."""
+
+from functools import cmp_to_key
+from typing import Dict, List, Optional, Sequence, Union
+
+import pyarrow as pa
+
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.options import Options
+from pypaimon.common.predicate import Predicate
+from pypaimon.globalindex.btree.btree_index_writer import (
+ BTREE_IDENTIFIER,
+ BTreeIndexWriter,
+)
+from pypaimon.globalindex.global_index_meta import GlobalIndexMeta
+from pypaimon.globalindex.key_serializer import create_serializer
+from pypaimon.index.index_file_meta import IndexFileMeta
+from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+from pypaimon.read.split import DataSplit
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.special_fields import SpecialFields
+from pypaimon.utils.range import Range
+from pypaimon.utils.range_helper import RangeHelper
+from pypaimon.write.commit_message import CommitMessage
+
+
+def create_global_index(
+ table,
+ index_column: Union[str, Sequence[str]],
+ index_type: str = BTREE_IDENTIFIER,
+ partition_filter: Optional[Predicate] = None,
+ partitions: Optional[Union[Dict[str, object], Sequence[Dict[str,
object]]]] = None,
+ options: Optional[Dict[str, object]] = None,
+) -> int:
+ """Build and commit global index files for a table.
+
+ Returns the number of index files added to the table snapshot.
+ """
+
+ builder = GlobalIndexBuilder(
+ table,
+ index_column,
+ index_type=index_type,
+ partition_filter=partition_filter,
+ partitions=partitions,
+ options=options,
+ )
+ messages = builder.build()
+ if not messages:
+ return 0
+
+ write_builder = table.new_batch_write_builder()
+ commit = write_builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ return sum(len(message.index_adds) for message in messages)
+
+
+class GlobalIndexBuilder:
+ """Small Python builder for sorted global indexes."""
+
+ def __init__(
+ self,
+ table,
+ index_column: Union[str, Sequence[str]],
+ index_type: str = BTREE_IDENTIFIER,
+ partition_filter: Optional[Predicate] = None,
+ partitions: Optional[Union[Dict[str, object], Sequence[Dict[str,
object]]]] = None,
+ options: Optional[Dict[str, object]] = None,
+ ):
+ self._table = table
+ self._index_columns = _normalize_index_columns(index_column)
+ self._index_type = index_type.lower().strip()
+ self._partition_filter = partition_filter
+ self._partitions = partitions
+ self._options = _merged_options(table, options)
+ self._core_options = CoreOptions(self._options)
+
+ if self._index_type != BTREE_IDENTIFIER:
+ raise ValueError(
+ "Python global index build currently supports only '%s', got
'%s'."
+ % (BTREE_IDENTIFIER, index_type)
+ )
+ if len(self._index_columns) != 1:
+ raise ValueError(
+ "Python '%s' global index build currently supports one column,
got %s."
+ % (BTREE_IDENTIFIER, self._index_columns)
+ )
+ if not self._table.options.row_tracking_enabled():
+ raise ValueError(
+ "Table '%s' must enable 'row-tracking.enabled=true' before "
+ "creating global index." % self._table.identifier
+ )
+ for column in self._index_columns:
+ if column not in self._table.field_dict:
+ raise ValueError(
+ "Column '%s' does not exist in table '%s'."
+ % (column, self._table.identifier)
+ )
+
+ def build(self) -> List[CommitMessage]:
+ read_builder = self._table.new_read_builder()
+ partition_filter = self._resolve_partition_filter(read_builder)
+ if partition_filter is not None:
+ read_builder = read_builder.with_partition_filter(partition_filter)
+
+ scan = read_builder.new_scan()
+ splits = scan.plan().splits()
+ if not splits:
+ return []
+
+ index_field = self._table.field_dict[self._index_columns[0]]
+ 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:
+ raise ValueError("sorted-index.records-per-range must be
positive.")
+
+ read_type = [index_field, SpecialFields.ROW_ID]
+ from pypaimon.read.table_read import TableRead
+
+ table_read = TableRead(
+ table=self._table,
+ predicate=None,
+ read_type=read_type,
+ )
+ index_path_factory =
self._table.path_factory().global_index_path_factory()
+ index_path = index_path_factory.global_index_root_path()
+
+ messages = []
+ for split in _split_by_contiguous_row_range(splits):
+ row_range = _calc_row_range(split)
+ table = table_read.to_arrow([split])
+ if table is None or table.num_rows == 0:
+ continue
+ rows = _extract_sorted_rows(
+ table,
+ self._index_columns[0],
+ SpecialFields.ROW_ID.name,
+ key_serializer,
+ )
+ if not rows:
+ 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,
+ )
+ for key, row_id in chunk:
+ writer.write(key, row_id - row_range.from_)
+ result_entries = writer.finish()
+ index_adds.extend(
+ _to_index_manifest_entries(
+ self._table,
+ split.partition,
+ row_range,
+ index_field.id,
+ self._index_type,
+ result_entries,
+ )
+ )
+ if index_adds:
+ messages.append(
+ CommitMessage(
+ partition=tuple(split.partition.values),
+ bucket=0,
+ new_files=[],
+ index_adds=index_adds,
+ )
+ )
+ return messages
+
+ def _resolve_partition_filter(self, read_builder) -> Optional[Predicate]:
+ if self._partition_filter is not None:
+ return self._partition_filter
+ if self._partitions is None:
+ return None
+
+ partitions = self._partitions
+ if isinstance(partitions, dict):
+ partitions = [partitions]
+
+ predicate_builder = read_builder.new_predicate_builder()
+ partition_predicates = []
+ for partition in partitions:
+ sub_predicates = []
+ for key, value in partition.items():
+ if key not in self._table.partition_keys:
+ raise ValueError(
+ "Partition spec key '%s' is not a partition column. "
+ "Partition keys are: %s"
+ % (key, list(self._table.partition_keys))
+ )
+ if value is None:
+ sub_predicates.append(predicate_builder.is_null(key))
+ else:
+ sub_predicates.append(predicate_builder.equal(key, value))
+ if sub_predicates:
+ partition_predicates.append(
+ predicate_builder.and_predicates(sub_predicates))
+ return predicate_builder.or_predicates(partition_predicates)
+
+
+def _normalize_index_columns(index_column: Union[str, Sequence[str]]) ->
List[str]:
+ if isinstance(index_column, str):
+ return [c.strip() for c in index_column.split(",") if c.strip()]
+ return [str(c).strip() for c in index_column if str(c).strip()]
+
+
+def _merged_options(table, options: Optional[Dict[str, object]]) -> Options:
+ merged = dict(table.options.options.to_map())
+ if options:
+ merged.update(options)
+ return Options(merged)
+
+
+def _calc_row_range(split) -> Range:
+ ranges = []
+ for file in split.files:
+ row_range = file.row_id_range()
+ if row_range is None:
+ raise ValueError(
+ "Cannot build global index because file '%s' has no row id
range."
+ % file.file_name
+ )
+ ranges.append(row_range)
+ if not ranges:
+ raise ValueError("Cannot build global index for an empty split.")
+ merged = Range.sort_and_merge_overlap(ranges, True, True)
+ return Range(merged[0].from_, merged[-1].to)
+
+
+def _split_by_contiguous_row_range(splits):
+ result = []
+ for split in splits:
+ result.extend(_split_one_by_contiguous_row_range(split))
+ return result
+
+
+def _split_one_by_contiguous_row_range(split):
+ for file in split.files:
+ if file.row_id_range() is None:
+ raise ValueError(
+ "Cannot build global index because file '%s' has no row id
range."
+ % file.file_name
+ )
+
+ range_helper = RangeHelper(lambda file: file.row_id_range())
+ ranges = range_helper.merge_overlapping_ranges(split.files)
+ if not ranges:
+ return []
+
+ result = []
+ current_segment = []
+ current_max_row_id = None
+ for range_files in ranges:
+ min_row_id = min(file.row_id_range().from_ for file in range_files)
+ max_row_id = max(file.row_id_range().to for file in range_files)
+ if (
+ not current_segment
+ or current_max_row_id is None
+ or current_max_row_id >= min_row_id - 1
+ ):
+ current_segment.extend(range_files)
+ current_max_row_id = max_row_id
+ else:
+ result.append(_copy_split_with_files(split, current_segment))
+ current_segment = list(range_files)
+ current_max_row_id = max_row_id
+
+ if current_segment:
+ result.append(_copy_split_with_files(split, current_segment))
+ return result
+
+
+def _copy_split_with_files(split, files):
+ data_deletion_files = None
+ if getattr(split, "data_deletion_files", None) is not None:
+ index_by_file = {id(file): i for i, file in enumerate(split.files)}
+ data_deletion_files = [
+ split.data_deletion_files[index_by_file[id(file)]]
+ for file in files
+ ]
+ return DataSplit(
+ files=list(files),
+ partition=split.partition,
+ bucket=split.bucket,
+ raw_convertible=getattr(split, "raw_convertible", False),
+ data_deletion_files=data_deletion_files,
+ )
+
+
+def _extract_sorted_rows(
+ table: pa.Table,
+ index_column: str,
+ row_id_column: str,
+ key_serializer,
+):
+ keys = table.column(index_column).to_pylist()
+ row_ids = table.column(row_id_column).to_pylist()
+ rows = []
+ for key, row_id in zip(keys, row_ids):
+ if row_id is None:
+ raise ValueError("Cannot build global index because _ROW_ID is
null.")
+ rows.append((key, int(row_id)))
+
+ comparator = key_serializer.create_comparator()
+
+ def compare(left, right):
+ left_key = left[0]
+ right_key = right[0]
+ if left_key is None and right_key is None:
+ return 0
+ if left_key is None:
+ return -1
+ if right_key is None:
+ return 1
+ return comparator(left_key, right_key)
+
+ return sorted(rows, key=cmp_to_key(compare))
+
+
+def _chunks(rows, size):
+ for start in range(0, len(rows), size):
+ yield rows[start:start + size]
+
+
+def _to_index_manifest_entries(
+ table,
+ partition: GenericRow,
+ row_range: Range,
+ index_field_id: int,
+ index_type: str,
+ result_entries,
+) -> List[IndexManifestEntry]:
+ path_factory = table.path_factory().global_index_path_factory()
+ entries = []
+ for result in result_entries:
+ file_path = path_factory.to_path(result.file_name)
+ file_size = table.file_io.get_file_size(file_path)
+ external_path = file_path if path_factory.is_external_path() else None
+ index_file = IndexFileMeta(
+ index_type=index_type,
+ file_name=result.file_name,
+ file_size=file_size,
+ row_count=result.row_count,
+ global_index_meta=GlobalIndexMeta(
+ row_range_start=row_range.from_,
+ row_range_end=row_range.to,
+ index_field_id=index_field_id,
+ extra_field_ids=None,
+ index_meta=result.meta,
+ ),
+ external_path=external_path,
+ )
+ entries.append(
+ IndexManifestEntry(
+ kind=0,
+ partition=partition,
+ bucket=0,
+ index_file=index_file,
+ )
+ )
+ return entries
diff --git a/paimon-python/pypaimon/globalindex/drop_global_index.py
b/paimon-python/pypaimon/globalindex/drop_global_index.py
new file mode 100644
index 0000000000..6be45499e8
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/drop_global_index.py
@@ -0,0 +1,177 @@
+# 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.
+
+"""Drop global index files from Python."""
+
+from typing import Dict, List, Optional, Sequence, Union
+
+from pypaimon.common.predicate import Predicate
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.globalindex.btree.btree_index_writer import BTREE_IDENTIFIER
+from pypaimon.index.index_file_handler import IndexFileHandler
+from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+from pypaimon.write.commit_message import CommitMessage
+
+
+def drop_global_index(
+ table,
+ index_column: Union[str, Sequence[str]],
+ index_type: str = BTREE_IDENTIFIER,
+ partition_filter: Optional[Predicate] = None,
+ partitions: Optional[Union[Dict[str, object], Sequence[Dict[str,
object]]]] = None,
+ dry_run: bool = False,
+) -> int:
+ """Drop global index metadata for a table.
+
+ Returns the number of matched index files. When ``dry_run`` is true, no
+ commit is created.
+ """
+
+ dropper = GlobalIndexDropper(
+ table,
+ index_column,
+ index_type=index_type,
+ partition_filter=partition_filter,
+ partitions=partitions,
+ )
+ messages = dropper.build()
+ dropped = sum(len(message.index_deletes) for message in messages)
+ if dry_run or not messages:
+ return dropped
+
+ write_builder = table.new_batch_write_builder()
+ commit = write_builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ return dropped
+
+
+class GlobalIndexDropper:
+ """Small Python dropper for global index manifest entries."""
+
+ def __init__(
+ self,
+ table,
+ index_column: Union[str, Sequence[str]],
+ index_type: str = BTREE_IDENTIFIER,
+ partition_filter: Optional[Predicate] = None,
+ partitions: Optional[Union[Dict[str, object], Sequence[Dict[str,
object]]]] = None,
+ ):
+ self._table = table
+ self._index_columns = _normalize_index_columns(index_column)
+ self._index_type = index_type.lower().strip()
+ self._partition_filter = partition_filter
+ self._partitions = partitions
+
+ if not self._index_columns:
+ raise ValueError("At least one index column is required.")
+ for column in self._index_columns:
+ if column not in self._table.field_dict:
+ raise ValueError(
+ "Column '%s' does not exist in table '%s'."
+ % (column, self._table.identifier)
+ )
+
+ def build(self) -> List[CommitMessage]:
+ snapshot = self._table.snapshot_manager().get_latest_snapshot()
+ if snapshot is None:
+ return []
+
+ target_field_ids = [
+ self._table.field_dict[column].id for column in self._index_columns
+ ]
+ partition_filter = self._resolve_partition_filter()
+
+ def should_delete(entry: IndexManifestEntry) -> bool:
+ index_file = entry.index_file
+ global_meta = index_file.global_index_meta
+ if index_file.index_type.lower().strip() != self._index_type or
global_meta is None:
+ return False
+ if _indexed_field_ids(global_meta) != target_field_ids:
+ return False
+ return partition_filter is None or
partition_filter.test(entry.partition)
+
+ entries = IndexFileHandler(self._table).scan(snapshot, should_delete)
+ if not entries:
+ return []
+
+ by_partition = {}
+ for entry in entries:
+ key = tuple(entry.partition.values)
+ by_partition.setdefault(key, []).append(
+ IndexManifestEntry(
+ kind=1,
+ partition=entry.partition,
+ bucket=entry.bucket,
+ index_file=entry.index_file,
+ )
+ )
+
+ return [
+ CommitMessage(
+ partition=partition,
+ bucket=0,
+ new_files=[],
+ index_deletes=deletes,
+ )
+ for partition, deletes in by_partition.items()
+ ]
+
+ def _resolve_partition_filter(self) -> Optional[Predicate]:
+ if self._partition_filter is not None:
+ return self._partition_filter
+ if self._partitions is None:
+ return None
+
+ partitions = self._partitions
+ if isinstance(partitions, dict):
+ partitions = [partitions]
+
+ predicate_builder = PredicateBuilder(self._table.partition_keys_fields)
+ partition_predicates = []
+ for partition in partitions:
+ sub_predicates = []
+ for key, value in partition.items():
+ if key not in self._table.partition_keys:
+ raise ValueError(
+ "Partition spec key '%s' is not a partition column. "
+ "Partition keys are: %s"
+ % (key, list(self._table.partition_keys))
+ )
+ if value is None:
+ sub_predicates.append(predicate_builder.is_null(key))
+ else:
+ sub_predicates.append(predicate_builder.equal(key, value))
+ if sub_predicates:
+ partition_predicates.append(
+ predicate_builder.and_predicates(sub_predicates))
+ return predicate_builder.or_predicates(partition_predicates)
+
+
+def _normalize_index_columns(index_column: Union[str, Sequence[str]]) ->
List[str]:
+ if isinstance(index_column, str):
+ return [c.strip() for c in index_column.split(",") if c.strip()]
+ return [str(c).strip() for c in index_column if str(c).strip()]
+
+
+def _indexed_field_ids(global_meta) -> List[int]:
+ field_ids = [global_meta.index_field_id]
+ if global_meta.extra_field_ids:
+ field_ids.extend(global_meta.extra_field_ids)
+ return field_ids
diff --git a/paimon-python/pypaimon/globalindex/key_serializer.py
b/paimon-python/pypaimon/globalindex/key_serializer.py
index 875d6b0bb1..49b0f27356 100644
--- a/paimon-python/pypaimon/globalindex/key_serializer.py
+++ b/paimon-python/pypaimon/globalindex/key_serializer.py
@@ -18,7 +18,11 @@
"""Key serializer for global indexes."""
from abc import ABC, abstractmethod
-from typing import Callable
+import datetime
+from decimal import Decimal, ROUND_HALF_UP
+import math
+import re
+from typing import Callable, Tuple
import struct
from pypaimon.schema.data_types import AtomicType
@@ -88,6 +92,32 @@ class LongSerializer(KeySerializer):
return compare
+class ByteSerializer(KeySerializer):
+ """Serializer for TINYINT type."""
+
+ def serialize(self, key: object) -> bytes:
+ return struct.pack('<b', int(key))
+
+ def deserialize(self, data: bytes) -> object:
+ return struct.unpack('<b', data)[0]
+
+ def create_comparator(self) -> Callable[[object, object], int]:
+ return _numeric_compare
+
+
+class ShortSerializer(KeySerializer):
+ """Serializer for SMALLINT type."""
+
+ def serialize(self, key: object) -> bytes:
+ return struct.pack('<h', int(key))
+
+ def deserialize(self, data: bytes) -> object:
+ return struct.unpack('<h', data)[0]
+
+ def create_comparator(self) -> Callable[[object, object], int]:
+ return _numeric_compare
+
+
class IntSerializer(KeySerializer):
"""Serializer for INT type."""
@@ -109,15 +139,300 @@ class IntSerializer(KeySerializer):
return compare
+class BooleanSerializer(KeySerializer):
+ """Serializer for BOOLEAN type."""
+
+ def serialize(self, key: object) -> bytes:
+ return b'\x01' if bool(key) else b'\x00'
+
+ def deserialize(self, data: bytes) -> object:
+ return data[0] == 1
+
+ def create_comparator(self) -> Callable[[object, object], int]:
+ def compare(a: object, b: object) -> int:
+ return _cmp(bool(a), bool(b))
+ return compare
+
+
+class FloatSerializer(KeySerializer):
+ """Serializer for FLOAT type."""
+
+ def serialize(self, key: object) -> bytes:
+ return struct.pack('<f', float(key))
+
+ def deserialize(self, data: bytes) -> object:
+ return struct.unpack('<f', data)[0]
+
+ def create_comparator(self) -> Callable[[object, object], int]:
+ return _float_compare
+
+
+class DoubleSerializer(KeySerializer):
+ """Serializer for DOUBLE type."""
+
+ def serialize(self, key: object) -> bytes:
+ return struct.pack('<d', float(key))
+
+ def deserialize(self, data: bytes) -> object:
+ return struct.unpack('<d', data)[0]
+
+ def create_comparator(self) -> Callable[[object, object], int]:
+ return _float_compare
+
+
+class DecimalSerializer(KeySerializer):
+ """Serializer for DECIMAL type."""
+
+ def __init__(self, precision: int, scale: int):
+ self._precision = precision
+ self._scale = scale
+
+ def serialize(self, key: object) -> bytes:
+ unscaled = _decimal_unscaled(key, self._scale)
+ if self._precision <= 18:
+ return struct.pack('<q', unscaled)
+ return _signed_big_endian_bytes(unscaled)
+
+ def deserialize(self, data: bytes) -> object:
+ if self._precision <= 18:
+ unscaled = struct.unpack('<q', data)[0]
+ else:
+ unscaled = int.from_bytes(data, byteorder='big', signed=True)
+ return Decimal(unscaled).scaleb(-self._scale)
+
+ def create_comparator(self) -> Callable[[object, object], int]:
+ def compare(a: object, b: object) -> int:
+ return _cmp(Decimal(a), Decimal(b))
+ return compare
+
+
+class DateSerializer(IntSerializer):
+ """Serializer for DATE type."""
+
+ def serialize(self, key: object) -> bytes:
+ return struct.pack('<i', _date_to_epoch_days(key))
+
+ def deserialize(self, data: bytes) -> object:
+ days = struct.unpack('<i', data)[0]
+ return datetime.date(1970, 1, 1) + datetime.timedelta(days=days)
+
+ def create_comparator(self) -> Callable[[object, object], int]:
+ def compare(a: object, b: object) -> int:
+ return _cmp(_date_to_epoch_days(a), _date_to_epoch_days(b))
+ return compare
+
+
+class TimeSerializer(IntSerializer):
+ """Serializer for TIME type."""
+
+ def serialize(self, key: object) -> bytes:
+ return struct.pack('<i', _time_to_millis(key))
+
+ def deserialize(self, data: bytes) -> object:
+ millis = struct.unpack('<i', data)[0]
+ seconds, millis = divmod(millis, 1000)
+ minutes, second = divmod(seconds, 60)
+ hour, minute = divmod(minutes, 60)
+ return datetime.time(hour, minute, second, millis * 1000)
+
+ def create_comparator(self) -> Callable[[object, object], int]:
+ def compare(a: object, b: object) -> int:
+ return _cmp(_time_to_millis(a), _time_to_millis(b))
+ return compare
+
+
+class TimestampSerializer(KeySerializer):
+ """Serializer for TIMESTAMP and TIMESTAMP WITH LOCAL TIME ZONE types."""
+
+ def __init__(self, precision: int):
+ self._precision = precision
+
+ def serialize(self, key: object) -> bytes:
+ millis, nano_of_millisecond = _timestamp_to_millis_nanos(
+ key, self._precision)
+ if self._precision <= 3:
+ return struct.pack('<q', millis)
+ return struct.pack('<q', millis) +
_write_var_len_int(nano_of_millisecond)
+
+ def deserialize(self, data: bytes) -> object:
+ millis = struct.unpack('<q', data[:8])[0]
+ nano_of_millisecond = 0
+ if self._precision > 3 and len(data) > 8:
+ nano_of_millisecond, _ = _read_var_len_int(data, 8)
+ total_micros = millis * 1000 + nano_of_millisecond // 1000
+ return (
+ datetime.datetime(1970, 1, 1)
+ + datetime.timedelta(microseconds=total_micros)
+ )
+
+ def create_comparator(self) -> Callable[[object, object], int]:
+ def compare(a: object, b: object) -> int:
+ return _cmp(
+ _timestamp_to_millis_nanos(a, self._precision),
+ _timestamp_to_millis_nanos(b, self._precision),
+ )
+ return compare
+
+
+def _cmp(a, b) -> int:
+ if a < b:
+ return -1
+ if a > b:
+ return 1
+ return 0
+
+
+def _numeric_compare(a: object, b: object) -> int:
+ return _cmp(int(a), int(b))
+
+
+def _float_compare(a: object, b: object) -> int:
+ left = float(a)
+ right = float(b)
+ left_nan = math.isnan(left)
+ right_nan = math.isnan(right)
+ if left_nan or right_nan:
+ if left_nan and right_nan:
+ return 0
+ return 1 if left_nan else -1
+ return _cmp(left, right)
+
+
+def _parse_decimal_params(type_name: str) -> Tuple[int, int]:
+ if type_name in ("DECIMAL", "NUMERIC"):
+ return 10, 0
+ match = re.fullmatch(r'(?:DECIMAL|NUMERIC)\((\d+),\s*(\d+)\)', type_name)
+ if match:
+ return int(match.group(1)), int(match.group(2))
+ match = re.fullmatch(r'(?:DECIMAL|NUMERIC)\((\d+)\)', type_name)
+ if match:
+ return int(match.group(1)), 0
+ raise ValueError(f"Invalid decimal type: {type_name}")
+
+
+def _parse_precision(type_name: str, default: int) -> int:
+ match = re.search(r'\((\d+)\)', type_name)
+ return int(match.group(1)) if match else default
+
+
+def _decimal_unscaled(value: object, scale: int) -> int:
+ decimal_value = value if isinstance(value, Decimal) else
Decimal(str(value))
+ quant = Decimal(1).scaleb(-scale)
+ rounded = decimal_value.quantize(quant, rounding=ROUND_HALF_UP)
+ return int(rounded.scaleb(scale))
+
+
+def _signed_big_endian_bytes(value: int) -> bytes:
+ length = max(1, (value.bit_length() + 8) // 8)
+ while True:
+ try:
+ encoded = value.to_bytes(length, byteorder='big', signed=True)
+ break
+ except OverflowError:
+ length += 1
+ while len(encoded) > 1:
+ if encoded[0] == 0x00 and encoded[1] < 0x80:
+ encoded = encoded[1:]
+ elif encoded[0] == 0xFF and encoded[1] >= 0x80:
+ encoded = encoded[1:]
+ else:
+ break
+ return encoded
+
+
+def _date_to_epoch_days(value: object) -> int:
+ if isinstance(value, datetime.datetime):
+ value = value.date()
+ if isinstance(value, datetime.date):
+ return (value - datetime.date(1970, 1, 1)).days
+ return int(value)
+
+
+def _time_to_millis(value: object) -> int:
+ if isinstance(value, datetime.time):
+ return (
+ ((value.hour * 60 + value.minute) * 60 + value.second) * 1000
+ + value.microsecond // 1000
+ )
+ return int(value)
+
+
+def _timestamp_to_millis_nanos(
+ value: object, precision: int = 6
+) -> Tuple[int, int]:
+ if hasattr(value, "get_millisecond") and hasattr(value,
"get_nano_of_millisecond"):
+ return int(value.get_millisecond()),
int(value.get_nano_of_millisecond())
+ if isinstance(value, datetime.datetime):
+ if value.tzinfo is None:
+ epoch = datetime.datetime(1970, 1, 1)
+ else:
+ epoch = datetime.datetime(1970, 1, 1, tzinfo=value.tzinfo)
+ delta = value - epoch
+ total_micros = (
+ delta.days * 86_400_000_000
+ + delta.seconds * 1_000_000
+ + delta.microseconds
+ )
+ millis, micros_in_milli = divmod(total_micros, 1000)
+ return millis, micros_in_milli * 1000
+ if precision <= 3:
+ return int(value), 0
+ micros = int(value)
+ millis, micros_in_milli = divmod(micros, 1000)
+ return millis, micros_in_milli * 1000
+
+
+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 _read_var_len_int(data: bytes, offset: int = 0):
+ shift = 0
+ result = 0
+ while True:
+ b = data[offset]
+ offset += 1
+ result |= (b & 0x7F) << shift
+ if (b & 0x80) == 0:
+ return result, offset
+ shift += 7
+
+
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'):
+ if type_name == 'BOOLEAN':
+ return BooleanSerializer()
+ if type_name == 'TINYINT':
+ return ByteSerializer()
+ if type_name == 'SMALLINT':
+ return ShortSerializer()
+ if type_name in ('CHAR', 'VARCHAR', 'STRING') or
type_name.startswith('CHAR') or type_name.startswith('VARCHAR'):
return StringSerializer()
+ if type_name in ('INT', 'INTEGER'):
+ return IntSerializer()
if type_name == 'BIGINT':
return LongSerializer()
- if type_name == 'INT':
- return IntSerializer()
+ if type_name == 'FLOAT':
+ return FloatSerializer()
+ if type_name == 'DOUBLE':
+ return DoubleSerializer()
+ if type_name.startswith('DECIMAL') or type_name.startswith('NUMERIC'):
+ precision, scale = _parse_decimal_params(type_name)
+ return DecimalSerializer(precision, scale)
+ if type_name == 'DATE':
+ return DateSerializer()
+ if type_name.startswith('TIME') and not type_name.startswith('TIMESTAMP'):
+ return TimeSerializer()
+ if type_name.startswith('TIMESTAMP'):
+ return TimestampSerializer(_parse_precision(type_name, 6))
raise ValueError(f"DataType: {data_type} is not supported by global index
now.")
diff --git a/paimon-python/pypaimon/globalindex/btree/__init__.py
b/paimon-python/pypaimon/globalindex/result_entry.py
similarity index 57%
copy from paimon-python/pypaimon/globalindex/btree/__init__.py
copy to paimon-python/pypaimon/globalindex/result_entry.py
index 17a6a30cbc..ba44a368a3 100644
--- a/paimon-python/pypaimon/globalindex/btree/__init__.py
+++ b/paimon-python/pypaimon/globalindex/result_entry.py
@@ -15,18 +15,14 @@
# specific language governing permissions and limitations
# under the License.
-"""B-tree index implementation for global index."""
+from dataclasses import dataclass
+from typing import Optional
-from pypaimon.globalindex.btree.btree_index_reader import BTreeIndexReader
-from pypaimon.globalindex.btree.btree_index_meta import BTreeIndexMeta
-from pypaimon.globalindex.btree.btree_file_meta_selector import
BTreeFileMetaSelector
-from pypaimon.globalindex.btree.key_serializer import KeySerializer
-from pypaimon.globalindex.btree.lazy_filtered_btree_reader import
LazyFilteredBTreeReader
-__all__ = [
- 'BTreeIndexReader',
- 'BTreeIndexMeta',
- 'BTreeFileMetaSelector',
- 'KeySerializer',
- 'LazyFilteredBTreeReader',
-]
+@dataclass(frozen=True)
+class ResultEntry:
+ """Write result metadata for one global index file."""
+
+ file_name: str
+ row_count: int
+ meta: Optional[bytes]
diff --git a/paimon-python/pypaimon/globalindex/sorted_index_file_meta.py
b/paimon-python/pypaimon/globalindex/sorted_index_file_meta.py
index be3e10b73a..e9d5d9b5ab 100644
--- a/paimon-python/pypaimon/globalindex/sorted_index_file_meta.py
+++ b/paimon-python/pypaimon/globalindex/sorted_index_file_meta.py
@@ -37,6 +37,29 @@ class SortedIndexFileMeta:
def only_nulls(self) -> bool:
return self.first_key is None and self.last_key is None
+ def serialize(self) -> bytes:
+ result = bytearray()
+ null_key_flags = 0
+
+ if self.first_key is None:
+ result.extend(struct.pack('<I', 0))
+ null_key_flags |= self.FIRST_KEY_IS_NULL
+ else:
+ result.extend(struct.pack('<I', len(self.first_key)))
+ result.extend(self.first_key)
+
+ if self.last_key is None:
+ result.extend(struct.pack('<I', 0))
+ null_key_flags |= self.LAST_KEY_IS_NULL
+ else:
+ result.extend(struct.pack('<I', len(self.last_key)))
+ result.extend(self.last_key)
+
+ result.extend(struct.pack('<B', 1 if self.has_nulls else 0))
+ result.extend(struct.pack('<B', self.FORMAT_VERSION_WITH_NULL_FLAGS))
+ result.extend(struct.pack('<B', null_key_flags))
+ return bytes(result)
+
@classmethod
def deserialize(cls, data: bytes) -> 'SortedIndexFileMeta':
offset = 0
diff --git a/paimon-python/pypaimon/manifest/index_manifest_file.py
b/paimon-python/pypaimon/manifest/index_manifest_file.py
index 62ca9ea16d..2d2f73e4da 100644
--- a/paimon-python/pypaimon/manifest/index_manifest_file.py
+++ b/paimon-python/pypaimon/manifest/index_manifest_file.py
@@ -36,7 +36,7 @@ _DELETION_VECTOR_META_SCHEMA = {
"name": "DeletionVectorMeta",
"fields": [
{"name": "f0", "type": "string"},
- {"name": "f1", "type": "long"},
+ {"name": "f1", "type": "int"},
{"name": "f2", "type": "int"},
{"name": "_CARDINALITY", "type": ["null", "long"], "default": None},
],
@@ -77,6 +77,8 @@ INDEX_MANIFEST_ENTRY_SCHEMA = {
}
_INDEX_ENTRY_VERSION = 1
+_ADD = 0
+_HASH_INDEX = "HASH"
class IndexManifestFile:
@@ -234,14 +236,24 @@ class IndexManifestFile:
previous_name: Optional[str],
deletes: List[IndexManifestEntry],
) -> Optional[str]:
- if not deletes:
+ return self.combine_changes(previous_name, [], deletes)
+
+ def combine_changes(
+ self,
+ previous_name: Optional[str],
+ adds: List[IndexManifestEntry],
+ deletes: List[IndexManifestEntry],
+ ) -> Optional[str]:
+ if not adds and not deletes:
return previous_name
previous = self.read(previous_name) if previous_name else []
delete_names = {e.index_file.file_name for e in deletes}
survivors = [e for e in previous if e.index_file.file_name not in
delete_names]
- if not survivors:
+ _validate_retained_global_index_files(survivors, adds)
+ combined = survivors + adds
+ if not combined:
return None
- return self.write(survivors)
+ return self.write(combined)
def write(self, entries: List[IndexManifestEntry]) -> str:
file_name =
f"{FileStorePathFactory.INDEX_MANIFEST_PREFIX}{uuid.uuid4()}"
@@ -293,3 +305,64 @@ class IndexManifestFile:
"_EXTERNAL_PATH": index_file.external_path,
"_GLOBAL_INDEX": global_index,
}
+
+
+def _validate_retained_global_index_files(
+ retained_entries: List[IndexManifestEntry],
+ added_entries: List[IndexManifestEntry],
+) -> None:
+ for retained in retained_entries:
+ retained_file = retained.index_file
+ retained_meta = retained_file.global_index_meta
+ if retained_meta is None or not
_is_global_index(retained_file.index_type):
+ continue
+
+ for added in added_entries:
+ added_file = added.index_file
+ added_meta = added_file.global_index_meta
+ if (
+ added.kind != _ADD
+ or added_meta is None
+ or added_file.index_type != retained_file.index_type
+ or retained_meta.index_field_id != added_meta.index_field_id
+ or _can_keep_existing_global_index(retained_meta, added_meta)
+ ):
+ continue
+
+ raise RuntimeError(
+ "Trying to add global index file %s of type %s for index field
%s"
+ " with row range [%s, %s], but previous file %s still exists"
+ " with overlapping row range [%s, %s]. Remove the previous
file first."
+ % (
+ added_file.file_name,
+ added_file.index_type,
+ added_meta.index_field_id,
+ added_meta.row_range_start,
+ added_meta.row_range_end,
+ retained_file.file_name,
+ retained_meta.row_range_start,
+ retained_meta.row_range_end,
+ )
+ )
+
+
+def _is_global_index(index_type: str) -> bool:
+ return index_type not in (IndexManifestFile.DELETION_VECTORS_INDEX,
_HASH_INDEX)
+
+
+def _can_keep_existing_global_index(retained_meta, added_meta) -> bool:
+ return (
+ _extra_field_ids(retained_meta) == _extra_field_ids(added_meta)
+ and not _row_ranges_intersect(retained_meta, added_meta)
+ )
+
+
+def _extra_field_ids(global_index_meta) -> Optional[List[int]]:
+ return global_index_meta.extra_field_ids or None
+
+
+def _row_ranges_intersect(left, right) -> bool:
+ return (
+ left.row_range_start <= right.row_range_end
+ and right.row_range_start <= left.row_range_end
+ )
diff --git a/paimon-python/pypaimon/table/file_store_table.py
b/paimon-python/pypaimon/table/file_store_table.py
index dd561eb55b..56e2b9518c 100644
--- a/paimon-python/pypaimon/table/file_store_table.py
+++ b/paimon-python/pypaimon/table/file_store_table.py
@@ -394,6 +394,7 @@ class FileStoreTable(Table):
external_path_strategy=self.options.data_file_external_paths_strategy(),
external_path_weights=self.options.data_file_external_paths_weights(),
index_file_in_data_file_dir=False,
+
global_index_external_path=self.options.global_index_external_path(),
)
def new_snapshot_commit(self):
@@ -449,6 +450,32 @@ class FileStoreTable(Table):
BatchVectorSearchBuilderImpl
return BatchVectorSearchBuilderImpl(self)
+ def create_global_index(self, index_column, index_type: str = "btree",
+ partition_filter=None, partitions=None,
+ options: Optional[dict] = None) -> int:
+ from pypaimon.globalindex.create_global_index import
create_global_index
+ return create_global_index(
+ self,
+ index_column,
+ index_type=index_type,
+ partition_filter=partition_filter,
+ partitions=partitions,
+ options=options,
+ )
+
+ def drop_global_index(self, index_column, index_type: str = "btree",
+ partition_filter=None, partitions=None,
+ dry_run: bool = False) -> int:
+ from pypaimon.globalindex.drop_global_index import drop_global_index
+ return drop_global_index(
+ self,
+ index_column,
+ index_type=index_type,
+ partition_filter=partition_filter,
+ partitions=partitions,
+ dry_run=dry_run,
+ )
+
def create_row_key_extractor(self) -> RowKeyExtractor:
bucket_mode = self.bucket_mode()
if bucket_mode == BucketMode.HASH_FIXED:
diff --git a/paimon-python/pypaimon/table/system/file_key_ranges_table.py
b/paimon-python/pypaimon/table/system/file_key_ranges_table.py
new file mode 100644
index 0000000000..67b60ff5bd
--- /dev/null
+++ b/paimon-python/pypaimon/table/system/file_key_ranges_table.py
@@ -0,0 +1,145 @@
+# 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.
+
+"""The ``$file_key_ranges`` system table."""
+
+from typing import List
+
+import pyarrow
+
+from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+from pypaimon.manifest.manifest_list_manager import ManifestListManager
+from pypaimon.schema.data_types import AtomicType, DataField, RowType
+from pypaimon.table.system.files_table import (
+ _render_key,
+ _render_partition,
+ _stringify_path,
+)
+from pypaimon.table.system.system_table import SystemTable
+
+
+TABLE_TYPE = RowType(False, [
+ DataField(0, "partition", AtomicType("STRING", nullable=True)),
+ DataField(1, "bucket", AtomicType("INT", nullable=False)),
+ DataField(2, "file_path", AtomicType("STRING", nullable=False)),
+ DataField(3, "file_format", AtomicType("STRING", nullable=False)),
+ DataField(4, "schema_id", AtomicType("BIGINT", nullable=False)),
+ DataField(5, "level", AtomicType("INT", nullable=False)),
+ DataField(6, "record_count", AtomicType("BIGINT", nullable=False)),
+ DataField(7, "file_size_in_bytes", AtomicType("BIGINT", nullable=False)),
+ DataField(8, "min_key", AtomicType("STRING", nullable=True)),
+ DataField(9, "max_key", AtomicType("STRING", nullable=True)),
+ DataField(10, "first_row_id", AtomicType("BIGINT", nullable=True)),
+])
+
+
+class FileKeyRangesTable(SystemTable):
+ """The ``$file_key_ranges`` system table."""
+
+ def system_table_name(self) -> str:
+ return "file_key_ranges"
+
+ def row_type(self) -> RowType:
+ return TABLE_TYPE
+
+ def primary_keys(self) -> List[str]:
+ return ["file_path"]
+
+ def _build_arrow_table(self) -> pyarrow.Table:
+ snapshot = self.base_table.snapshot_manager().get_latest_snapshot()
+ if snapshot is None:
+ return _empty_table()
+
+ manifest_list_manager = ManifestListManager(self.base_table)
+ manifest_files = manifest_list_manager.read_all(snapshot)
+ manifest_file_manager = ManifestFileManager(self.base_table)
+ entries = manifest_file_manager.read_entries_parallel(
+ manifest_files, drop_stats=False)
+
+ file_format = self.base_table.options.file_format()
+ path_factory = self.base_table.path_factory()
+ rows = {
+ "partition": [],
+ "bucket": [],
+ "file_path": [],
+ "file_format": [],
+ "schema_id": [],
+ "level": [],
+ "record_count": [],
+ "file_size_in_bytes": [],
+ "min_key": [],
+ "max_key": [],
+ "first_row_id": [],
+ }
+
+ for entry in entries:
+ meta = entry.file
+ bucket_path = path_factory.bucket_path(
+ tuple(entry.partition.values), int(entry.bucket))
+ rows["partition"].append(_render_partition(entry.partition))
+ rows["bucket"].append(int(entry.bucket))
+ rows["file_path"].append(_stringify_path(
+ meta.external_path
+ or meta.file_path
+ or "%s/%s" % (bucket_path, meta.file_name)))
+ rows["file_format"].append(file_format)
+ rows["schema_id"].append(int(meta.schema_id))
+ rows["level"].append(int(meta.level))
+ rows["record_count"].append(int(meta.row_count))
+ rows["file_size_in_bytes"].append(int(meta.file_size))
+ rows["min_key"].append(_render_key(meta.min_key))
+ rows["max_key"].append(_render_key(meta.max_key))
+ rows["first_row_id"].append(
+ None if meta.first_row_id is None
+ else int(meta.first_row_id))
+
+ return pyarrow.table({
+ "partition": pyarrow.array(
+ rows["partition"], type=pyarrow.string()),
+ "bucket": pyarrow.array(rows["bucket"], type=pyarrow.int32()),
+ "file_path": pyarrow.array(
+ rows["file_path"], type=pyarrow.string()),
+ "file_format": pyarrow.array(
+ rows["file_format"], type=pyarrow.string()),
+ "schema_id": pyarrow.array(
+ rows["schema_id"], type=pyarrow.int64()),
+ "level": pyarrow.array(rows["level"], type=pyarrow.int32()),
+ "record_count": pyarrow.array(
+ rows["record_count"], type=pyarrow.int64()),
+ "file_size_in_bytes": pyarrow.array(
+ rows["file_size_in_bytes"], type=pyarrow.int64()),
+ "min_key": pyarrow.array(rows["min_key"], type=pyarrow.string()),
+ "max_key": pyarrow.array(rows["max_key"], type=pyarrow.string()),
+ "first_row_id": pyarrow.array(
+ rows["first_row_id"], type=pyarrow.int64()),
+ })
+
+
+def _empty_table() -> pyarrow.Table:
+ return pyarrow.table({
+ "partition": pyarrow.array([], type=pyarrow.string()),
+ "bucket": pyarrow.array([], type=pyarrow.int32()),
+ "file_path": pyarrow.array([], type=pyarrow.string()),
+ "file_format": pyarrow.array([], type=pyarrow.string()),
+ "schema_id": pyarrow.array([], type=pyarrow.int64()),
+ "level": pyarrow.array([], type=pyarrow.int32()),
+ "record_count": pyarrow.array([], type=pyarrow.int64()),
+ "file_size_in_bytes": pyarrow.array([], type=pyarrow.int64()),
+ "min_key": pyarrow.array([], type=pyarrow.string()),
+ "max_key": pyarrow.array([], type=pyarrow.string()),
+ "first_row_id": pyarrow.array([], type=pyarrow.int64()),
+ })
diff --git a/paimon-python/pypaimon/table/system/system_table_loader.py
b/paimon-python/pypaimon/table/system/system_table_loader.py
index 72b758947d..c2f50649a5 100644
--- a/paimon-python/pypaimon/table/system/system_table_loader.py
+++ b/paimon-python/pypaimon/table/system/system_table_loader.py
@@ -24,8 +24,7 @@ new module.
The following short names are intentionally not registered here yet:
audit_log, binlog, read_optimized, consumers, statistics,
- aggregation_fields, file_key_ranges, table_indexes,
- row_tracking, all_tables, all_partitions, all_table_options,
+ aggregation_fields, row_tracking, all_tables, all_partitions,
all_table_options,
catalog_options
"""
@@ -47,6 +46,8 @@ SYSTEM_TABLES: Tuple[str, ...] = (
"buckets",
"tags",
"branches",
+ "file_key_ranges",
+ "table_indexes",
)
@@ -70,6 +71,10 @@ SYSTEM_TABLE_LOADERS: Dict[str, Callable[...,
"SystemTable"]] = {
"buckets": _lazy("pypaimon.table.system.buckets_table", "BucketsTable"),
"tags": _lazy("pypaimon.table.system.tags_table", "TagsTable"),
"branches": _lazy("pypaimon.table.system.branches_table", "BranchesTable"),
+ "file_key_ranges": _lazy(
+ "pypaimon.table.system.file_key_ranges_table", "FileKeyRangesTable"),
+ "table_indexes": _lazy(
+ "pypaimon.table.system.table_indexes_table", "TableIndexesTable"),
}
diff --git a/paimon-python/pypaimon/table/system/table_indexes_table.py
b/paimon-python/pypaimon/table/system/table_indexes_table.py
new file mode 100644
index 0000000000..4bf2103f0b
--- /dev/null
+++ b/paimon-python/pypaimon/table/system/table_indexes_table.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.
+
+"""The ``$table_indexes`` system table."""
+
+from typing import Dict, List, Optional
+
+import pyarrow
+
+from pypaimon.index.index_file_handler import IndexFileHandler
+from pypaimon.schema.data_types import (ArrayType, AtomicType, DataField,
+ RowType)
+from pypaimon.table.system.files_table import _render_partition
+from pypaimon.table.system.system_table import SystemTable
+
+
+_DV_RANGE_ROW_TYPE = RowType(False, [
+ DataField(0, "f0", AtomicType("STRING", nullable=False)),
+ DataField(1, "f1", AtomicType("INT", nullable=False)),
+ DataField(2, "f2", AtomicType("INT", nullable=False)),
+ DataField(3, "_CARDINALITY", AtomicType("BIGINT", nullable=True)),
+])
+_DV_RANGES_TYPE = pyarrow.list_(pyarrow.struct([
+ pyarrow.field("f0", pyarrow.string(), nullable=False),
+ pyarrow.field("f1", pyarrow.int32(), nullable=False),
+ pyarrow.field("f2", pyarrow.int32(), nullable=False),
+ pyarrow.field("_CARDINALITY", pyarrow.int64(), nullable=True),
+]))
+
+TABLE_TYPE = RowType(False, [
+ DataField(0, "partition", AtomicType("STRING", nullable=True)),
+ DataField(1, "bucket", AtomicType("INT", nullable=False)),
+ DataField(2, "index_type", AtomicType("STRING", nullable=False)),
+ DataField(3, "file_name", AtomicType("STRING", nullable=False)),
+ DataField(4, "file_size", AtomicType("BIGINT", nullable=False)),
+ DataField(5, "row_count", AtomicType("BIGINT", nullable=False)),
+ DataField(
+ 6,
+ "dv_ranges",
+ ArrayType(nullable=True,
+ element_type=_DV_RANGE_ROW_TYPE)),
+ DataField(7, "row_range_start", AtomicType("BIGINT", nullable=True)),
+ DataField(8, "row_range_end", AtomicType("BIGINT", nullable=True)),
+ DataField(9, "index_field_id", AtomicType("INT", nullable=True)),
+ DataField(10, "index_field_name", AtomicType("STRING", nullable=True)),
+])
+
+
+class TableIndexesTable(SystemTable):
+ """The ``$table_indexes`` system table."""
+
+ def system_table_name(self) -> str:
+ return "table_indexes"
+
+ def row_type(self) -> RowType:
+ return TABLE_TYPE
+
+ def primary_keys(self) -> List[str]:
+ return ["file_name"]
+
+ def _build_arrow_table(self) -> pyarrow.Table:
+ snapshot = self.base_table.snapshot_manager().get_latest_snapshot()
+ entries = IndexFileHandler(self.base_table).scan(snapshot)
+
+ rows = {
+ "partition": [],
+ "bucket": [],
+ "index_type": [],
+ "file_name": [],
+ "file_size": [],
+ "row_count": [],
+ "dv_ranges": [],
+ "row_range_start": [],
+ "row_range_end": [],
+ "index_field_id": [],
+ "index_field_name": [],
+ }
+
+ for entry in entries:
+ index_file = entry.index_file
+ global_meta = index_file.global_index_meta
+ rows["partition"].append(_render_partition(entry.partition))
+ rows["bucket"].append(int(entry.bucket))
+ rows["index_type"].append(index_file.index_type)
+ rows["file_name"].append(index_file.file_name)
+ rows["file_size"].append(int(index_file.file_size))
+ rows["row_count"].append(int(index_file.row_count))
+ rows["dv_ranges"].append(_render_dv_ranges(index_file.dv_ranges))
+ rows["row_range_start"].append(
+ None if global_meta is None
+ else int(global_meta.row_range_start))
+ rows["row_range_end"].append(
+ None if global_meta is None
+ else int(global_meta.row_range_end))
+ rows["index_field_id"].append(
+ None if global_meta is None
+ else int(global_meta.index_field_id))
+ rows["index_field_name"].append(
+ None if global_meta is None
+ else _index_field_names(self.base_table, global_meta))
+
+ return pyarrow.table({
+ "partition": pyarrow.array(
+ rows["partition"], type=pyarrow.string()),
+ "bucket": pyarrow.array(rows["bucket"], type=pyarrow.int32()),
+ "index_type": pyarrow.array(
+ rows["index_type"], type=pyarrow.string()),
+ "file_name": pyarrow.array(
+ rows["file_name"], type=pyarrow.string()),
+ "file_size": pyarrow.array(
+ rows["file_size"], type=pyarrow.int64()),
+ "row_count": pyarrow.array(
+ rows["row_count"], type=pyarrow.int64()),
+ "dv_ranges": pyarrow.array(
+ rows["dv_ranges"], type=_DV_RANGES_TYPE),
+ "row_range_start": pyarrow.array(
+ rows["row_range_start"], type=pyarrow.int64()),
+ "row_range_end": pyarrow.array(
+ rows["row_range_end"], type=pyarrow.int64()),
+ "index_field_id": pyarrow.array(
+ rows["index_field_id"], type=pyarrow.int32()),
+ "index_field_name": pyarrow.array(
+ rows["index_field_name"], type=pyarrow.string()),
+ })
+
+
+def _render_dv_ranges(dv_ranges) -> Optional[List[Dict[str, object]]]:
+ if not dv_ranges:
+ return None
+ rendered = []
+ for meta in dv_ranges.values():
+ rendered.append({
+ "f0": meta.data_file_name,
+ "f1": int(meta.offset),
+ "f2": int(meta.length),
+ "_CARDINALITY": (
+ None if meta.cardinality is None else int(meta.cardinality)
+ ),
+ })
+ return rendered
+
+
+def _index_field_names(table, global_meta) -> Optional[str]:
+ field_by_id = {field.id: field.name for field in table.fields}
+ field_ids = [global_meta.index_field_id]
+ if global_meta.extra_field_ids:
+ field_ids.extend(global_meta.extra_field_ids)
+
+ names = []
+ for field_id in field_ids:
+ name = field_by_id.get(field_id)
+ if name is None:
+ return None
+ names.append(name)
+ return ",".join(names)
diff --git a/paimon-python/pypaimon/tests/global_index_build_test.py
b/paimon-python/pypaimon/tests/global_index_build_test.py
new file mode 100644
index 0000000000..15fd3bfd39
--- /dev/null
+++ b/paimon-python/pypaimon/tests/global_index_build_test.py
@@ -0,0 +1,292 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import unittest
+from datetime import date, datetime
+from decimal import Decimal
+import os
+
+import pyarrow as pa
+
+from pypaimon.globalindex.create_global_index import (
+ _split_one_by_contiguous_row_range,
+)
+from pypaimon.globalindex.key_serializer import create_serializer
+from pypaimon.globalindex.global_index_scanner import GlobalIndexScanner
+from pypaimon.index.index_file_handler import IndexFileHandler
+from pypaimon.schema.data_types import ArrayType, AtomicType, RowType
+from pypaimon.tests.data_evolution_test_helpers import (
+ BatchModeMixin,
+ DataEvolutionTestBase,
+)
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.utils.range import Range
+
+
+class _FakeFile:
+
+ def __init__(self, file_name, first_row_id, row_count):
+ self.file_name = file_name
+ self.first_row_id = first_row_id
+ self.row_count = row_count
+
+ def row_id_range(self):
+ return Range(self.first_row_id,
+ self.first_row_id + self.row_count - 1)
+
+
+class _FakeSplit:
+
+ def __init__(self, files):
+ self.files = files
+ self.partition = GenericRow([], [])
+ self.bucket = 0
+ self.raw_convertible = False
+
+
+class GlobalIndexBuildTest(
+ BatchModeMixin, DataEvolutionTestBase, unittest.TestCase):
+
+ table_options = {
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'global-index.enabled': 'true',
+ 'bucket': '-1',
+ 'file.format': 'parquet',
+ }
+
+ def test_create_btree_global_index_from_python(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.table(
+ {
+ 'id': [3, 1, 2, 2],
+ 'name': ['c', 'a', 'b1', 'b2'],
+ 'age': [30, 10, 20, 21],
+ 'city': ['z', 'x', 'y', 'y2'],
+ },
+ schema=self.pa_schema,
+ ))
+
+ added = table.create_global_index(
+ 'id',
+ options={'sorted-index.records-per-range': '2'},
+ )
+
+ self.assertEqual(2, added)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ self.assertIsNotNone(snapshot.index_manifest)
+
+ entries = IndexFileHandler(table).scan(snapshot)
+ self.assertEqual(2, len(entries))
+ self.assertEqual({'btree'}, {e.index_file.index_type for e in entries})
+ self.assertEqual({0}, {e.index_file.global_index_meta.row_range_start
for e in entries})
+ self.assertEqual({3}, {e.index_file.global_index_meta.row_range_end
for e in entries})
+
+ read_builder = table.new_read_builder()
+ predicate = read_builder.new_predicate_builder().equal('id', 2)
+ with GlobalIndexScanner.create(
+ table,
+ predicate=predicate,
+ snapshot=snapshot) as scanner:
+ result = scanner.scan(predicate)
+
+ self.assertEqual(
+ [Range(2, 3)],
+ result.results().to_range_list(),
+ )
+
+ table_name = table.identifier.get_full_name()
+ table_indexes = self.catalog.get_table(table_name + '$table_indexes')
+ index_read_builder = table_indexes.new_read_builder().with_projection([
+ 'index_type',
+ 'index_field_name',
+ 'row_range_start',
+ 'row_range_end',
+ ])
+ index_table = index_read_builder.new_read().to_arrow(
+ index_read_builder.new_scan().plan().splits())
+ self.assertEqual(2, index_table.num_rows)
+ self.assertEqual(['btree', 'btree'],
+ index_table.column('index_type').to_pylist())
+ self.assertEqual(['id', 'id'],
+ index_table.column('index_field_name').to_pylist())
+ self.assertEqual([0, 0],
+ index_table.column('row_range_start').to_pylist())
+ self.assertEqual([3, 3],
+ index_table.column('row_range_end').to_pylist())
+
+ key_ranges = self.catalog.get_table(table_name + '$file_key_ranges')
+ range_read_builder = key_ranges.new_read_builder().with_projection([
+ 'file_path',
+ 'record_count',
+ 'first_row_id',
+ ])
+ range_table = range_read_builder.new_read().to_arrow(
+ range_read_builder.new_scan().plan().splits())
+ self.assertEqual(1, range_table.num_rows)
+ file_path = range_table.column('file_path').to_pylist()[0]
+ self.assertIn('/bucket-0/', file_path)
+ self.assertEqual([4], range_table.column('record_count').to_pylist())
+ self.assertEqual([0], range_table.column('first_row_id').to_pylist())
+
+ dv_ranges_type = table_indexes.row_type().fields[6].type
+ self.assertIsInstance(dv_ranges_type, ArrayType)
+ self.assertIsInstance(dv_ranges_type.element, RowType)
+ self.assertEqual(
+ ['f0', 'f1', 'f2', '_CARDINALITY'],
+ [field.name for field in dv_ranges_type.element.fields],
+ )
+
+ self.assertEqual(2, table.drop_global_index('id', dry_run=True))
+ self.assertEqual(2, len(IndexFileHandler(table).scan(
+ table.snapshot_manager().get_latest_snapshot())))
+
+ self.assertEqual(2, table.drop_global_index('id'))
+ latest_snapshot = table.snapshot_manager().get_latest_snapshot()
+ self.assertEqual([], IndexFileHandler(table).scan(latest_snapshot))
+
+ index_read_builder = table_indexes.new_read_builder()
+ index_table = index_read_builder.new_read().to_arrow(
+ index_read_builder.new_scan().plan().splits())
+ self.assertEqual(0, index_table.num_rows)
+
+ def test_create_global_index_uses_external_path(self):
+ external_root = 'file://%s' % os.path.join(
+ self.tempdir, 'global-index-external')
+ options = dict(self.table_options)
+ options['global-index.external-path'] = external_root
+ table = self._create_table(options=options)
+ self._write_arrow(table, pa.table(
+ {
+ 'id': [1, 2],
+ 'name': ['a', 'b'],
+ 'age': [10, 20],
+ 'city': ['x', 'y'],
+ },
+ schema=self.pa_schema,
+ ))
+
+ self.assertEqual(1, table.create_global_index('id'))
+
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ entries = IndexFileHandler(table).scan(snapshot)
+ self.assertEqual(1, len(entries))
+ external_path = entries[0].index_file.external_path
+ self.assertIsNotNone(external_path)
+ self.assertTrue(external_path.startswith(external_root + '/'))
+ self.assertTrue(table.file_io.exists(external_path))
+
+ def test_create_global_index_rejects_overlapping_existing_range(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.table(
+ {
+ 'id': [3, 1, 2, 2],
+ 'name': ['c', 'a', 'b1', 'b2'],
+ 'age': [30, 10, 20, 21],
+ 'city': ['z', 'x', 'y', 'y2'],
+ },
+ schema=self.pa_schema,
+ ))
+ options = {'sorted-index.records-per-range': '2'}
+
+ self.assertEqual(2, table.create_global_index('id', options=options))
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ self.assertEqual(2, len(IndexFileHandler(table).scan(snapshot)))
+
+ with self.assertRaisesRegex(RuntimeError, 'overlapping row range'):
+ table.create_global_index('id', options=options)
+
+ latest_snapshot = table.snapshot_manager().get_latest_snapshot()
+ self.assertEqual(2, len(IndexFileHandler(table).scan(latest_snapshot)))
+
+ def test_create_btree_global_index_for_java_scalar_types(self):
+ schema = pa.schema([
+ ('flag', pa.bool_()),
+ ('amount', pa.decimal128(10, 2)),
+ ('dt', pa.date32()),
+ ('ts', pa.timestamp('us')),
+ ('payload', pa.string()),
+ ])
+ table = self._create_table(pa_schema=schema,
options=self.table_options)
+ self._write_arrow(table, pa.table(
+ {
+ 'flag': [True, False, True],
+ 'amount': [
+ Decimal('10.25'), Decimal('20.50'), Decimal('30.75')],
+ 'dt': [
+ date(2026, 6, 18),
+ date(2026, 6, 19),
+ date(2026, 6, 20),
+ ],
+ 'ts': [
+ datetime(2026, 6, 18, 10, 0, 0, 123456),
+ datetime(2026, 6, 19, 10, 0, 0, 123456),
+ datetime(2026, 6, 20, 10, 0, 0, 123456),
+ ],
+ 'payload': ['a', 'b', 'c'],
+ },
+ schema=schema,
+ ))
+
+ for column in ['flag', 'amount', 'dt', 'ts']:
+ self.assertEqual(1, table.create_global_index(column))
+
+ def test_split_by_contiguous_row_range_matches_java_builder(self):
+ split = _FakeSplit([
+ _FakeFile('a', 0, 2),
+ _FakeFile('b', 4, 2),
+ _FakeFile('c', 6, 1),
+ _FakeFile('d', 10, 1),
+ ])
+
+ splits = _split_one_by_contiguous_row_range(split)
+
+ self.assertEqual(
+ [['a'], ['b', 'c'], ['d']],
+ [[file.file_name for file in s.files] for s in splits],
+ )
+
+ def test_java_scalar_key_serializers_round_trip(self):
+ cases = [
+ ('BOOLEAN', True),
+ ('TINYINT', -7),
+ ('SMALLINT', 1024),
+ ('INT', 42),
+ ('BIGINT', 1234567890123),
+ ('FLOAT', 1.25),
+ ('DOUBLE', 3.14159),
+ ('DECIMAL(20, 5)', Decimal('-1234567890123.45678')),
+ ('DATE', date(2026, 6, 18)),
+ ('TIME(3)', datetime(2026, 6, 18, 1, 2, 3, 456000).time()),
+ ('TIMESTAMP(6)', datetime(2026, 6, 18, 1, 2, 3, 456789)),
+ ('VARCHAR(16)', 'abc'),
+ ]
+
+ for type_name, value in cases:
+ with self.subTest(type_name=type_name):
+ serializer = create_serializer(AtomicType(type_name))
+ actual = serializer.deserialize(serializer.serialize(value))
+ if type_name == 'FLOAT':
+ self.assertAlmostEqual(value, actual, places=6)
+ elif type_name == 'DOUBLE':
+ self.assertAlmostEqual(value, actual, places=12)
+ else:
+ self.assertEqual(value, actual)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/system/system_table_loader_test.py
b/paimon-python/pypaimon/tests/system/system_table_loader_test.py
index 07b959a2c6..3553d647a2 100644
--- a/paimon-python/pypaimon/tests/system/system_table_loader_test.py
+++ b/paimon-python/pypaimon/tests/system/system_table_loader_test.py
@@ -31,6 +31,8 @@ _EXPECTED_SYSTEM_TABLES = (
"buckets",
"tags",
"branches",
+ "file_key_ranges",
+ "table_indexes",
)
# Short names recognised by the Paimon catalog that this loader does
@@ -42,8 +44,6 @@ _UNREGISTERED_NAMES = {
"consumers",
"statistics",
"aggregation_fields",
- "file_key_ranges",
- "table_indexes",
"row_tracking",
"all_tables",
"all_partitions",
diff --git a/paimon-python/pypaimon/utils/file_store_path_factory.py
b/paimon-python/pypaimon/utils/file_store_path_factory.py
index 3816a2ce29..a98ddf00ce 100644
--- a/paimon-python/pypaimon/utils/file_store_path_factory.py
+++ b/paimon-python/pypaimon/utils/file_store_path_factory.py
@@ -58,6 +58,7 @@ class FileStorePathFactory:
external_path_strategy: str = "round-robin",
external_path_weights: Optional[List[int]] = None,
index_file_in_data_file_dir: bool = False,
+ global_index_external_path: Optional[str] = None,
):
self._root = root.rstrip('/')
self.partition_keys = partition_keys
@@ -73,6 +74,9 @@ class FileStorePathFactory:
self.external_path_weights = external_path_weights
self.index_file_in_data_file_dir = index_file_in_data_file_dir
self.legacy_partition_name = legacy_partition_name
+ self.global_index_external_path = (
+ global_index_external_path.rstrip('/')
+ if global_index_external_path else None)
def root(self) -> str:
return self._root
@@ -83,6 +87,9 @@ class FileStorePathFactory:
def index_path(self) -> str:
return f"{self._root}/{self.INDEX_PATH}"
+ def global_index_root_path(self) -> str:
+ return self.global_index_external_path or self.index_path()
+
def statistics_path(self) -> str:
return f"{self._root}/{self.STATISTICS_PATH}"
@@ -136,22 +143,37 @@ class FileStorePathFactory:
)
def global_index_path_factory(self) -> 'IndexPathFactory':
- return IndexPathFactory(self.index_path())
+ return IndexPathFactory(
+ self.index_path(),
+ self.global_index_root_path(),
+ self.global_index_external_path is not None,
+ )
class IndexPathFactory:
- def __init__(self, index_path: str):
+ def __init__(
+ self,
+ index_path: str,
+ global_index_root_path: Optional[str] = None,
+ external_path: bool = False,
+ ):
self._index_path = index_path
+ self._global_index_root_path = global_index_root_path or index_path
+ self._external_path = external_path
self._file_count = 0
def index_path(self) -> str:
- """Return the base index path."""
+ """Return the table index path used as a read fallback."""
return self._index_path
+ def global_index_root_path(self) -> str:
+ """Return the root path for newly written global index files."""
+ return self._global_index_root_path
+
def to_path(self, file_name: str) -> str:
"""Convert a file name to a full path."""
- return f"{self._index_path}/{file_name}"
+ return f"{self._global_index_root_path}/{file_name}"
def new_path(self, prefix: str = "index-") -> str:
"""Create a new unique index file path."""
@@ -162,4 +184,4 @@ class IndexPathFactory:
def is_external_path(self) -> bool:
"""Return whether this is an external path."""
- return False
+ return self._external_path
diff --git a/paimon-python/pypaimon/write/commit_message.py
b/paimon-python/pypaimon/write/commit_message.py
index a8cc193727..99e5e68f8c 100644
--- a/paimon-python/pypaimon/write/commit_message.py
+++ b/paimon-python/pypaimon/write/commit_message.py
@@ -30,8 +30,14 @@ class CommitMessage:
bucket: int
new_files: List[DataFileMeta]
check_from_snapshot: Optional[int] = -1
+ index_adds: List['IndexManifestEntry'] = field(default_factory=list)
index_deletes: List['IndexManifestEntry'] = field(default_factory=list)
changelog_files: List[DataFileMeta] = field(default_factory=list)
def is_empty(self):
- return not self.new_files and not self.index_deletes and not
self.changelog_files
+ return (
+ not self.new_files
+ and not self.index_adds
+ and not self.index_deletes
+ and not self.changelog_files
+ )
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index fb90f1aa90..0304178195 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -156,8 +156,10 @@ class FileStoreCommit:
len(commit_entries), len(changelog_entries))
index_deletes = []
+ index_adds = []
for msg in commit_messages:
index_deletes.extend(msg.index_deletes)
+ index_adds.extend(msg.index_adds)
if not index_deletes:
from pypaimon.write.global_index_update_checker import (
@@ -197,7 +199,8 @@ class FileStoreCommit:
changelog_entries=changelog_entries,
detect_conflicts=detect_conflicts,
allow_rollback=allow_rollback,
- index_deletes=index_deletes)
+ index_deletes=index_deletes,
+ index_adds=index_adds)
def overwrite(self, overwrite_partition, commit_messages:
List[CommitMessage], commit_identifier: int):
"""Commit the given commit messages in overwrite mode."""
@@ -286,7 +289,8 @@ class FileStoreCommit:
)
def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan,
- detect_conflicts=False, allow_rollback=False,
index_deletes=None, changelog_entries=None):
+ detect_conflicts=False, allow_rollback=False,
index_deletes=None,
+ index_adds=None, changelog_entries=None):
retry_count = 0
retry_result = None
@@ -297,7 +301,7 @@ class FileStoreCommit:
# No entries to commit (e.g. drop_partitions with no matching
data): skip commit
# to avoid creating manifest/snapshot with empty partition_stats
(causes read errors).
- if not commit_entries and not index_deletes:
+ if not commit_entries and not index_deletes and not index_adds:
break
result = self._try_commit_once(
@@ -310,6 +314,7 @@ class FileStoreCommit:
detect_conflicts=detect_conflicts,
allow_rollback=allow_rollback,
index_deletes=index_deletes,
+ index_adds=index_adds,
)
if result.is_success():
@@ -364,7 +369,8 @@ class FileStoreCommit:
latest_snapshot: Optional[Snapshot],
detect_conflicts: bool = False,
allow_rollback: bool = False,
- index_deletes=None) -> CommitResult:
+ index_deletes=None,
+ index_adds=None) -> CommitResult:
start_millis = int(time.time() * 1000)
if self._is_duplicate_commit(retry_result, latest_snapshot,
commit_identifier, commit_kind):
return SuccessResult()
@@ -466,11 +472,11 @@ class FileStoreCommit:
index_manifest = None
if latest_snapshot and commit_kind == "APPEND":
index_manifest = latest_snapshot.index_manifest
- if index_deletes:
+ if index_deletes or index_adds:
from pypaimon.manifest.index_manifest_file import
IndexManifestFile
previous_index_manifest = index_manifest
- index_manifest = IndexManifestFile(self.table).combine_deletes(
- previous_index_manifest, index_deletes)
+ index_manifest = IndexManifestFile(self.table).combine_changes(
+ previous_index_manifest, index_adds or [], index_deletes
or [])
if index_manifest != previous_index_manifest:
new_index_manifest = index_manifest
@@ -696,6 +702,19 @@ class FileStoreCommit:
except Exception as e:
path_to_delete = file.external_path if file.external_path
else file.file_path
logger.warning(f"Failed to clean up file {path_to_delete}
during abort: {e}")
+ for entry in message.index_adds:
+ try:
+ file_name = entry.index_file.file_name
+ index_path = (
+ entry.index_file.external_path
+ or self.table.path_factory()
+ .global_index_path_factory()
+ .to_path(file_name)
+ )
+ self.table.file_io.delete_quietly(index_path)
+ except Exception as e:
+ logger.warning(
+ f"Failed to clean up index file
{entry.index_file.file_name} during abort: {e}")
def close(self):
"""Close the FileStoreCommit and release resources."""
