[python] Complete native index lifecycle and explicit search plans

paimon-python/pypaimon/globalindex/build_plan.py

@@ -318,17 +318,11 @@ def split_one_by_contiguous_row_range(split):
 
 
 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
-        ]
+    # Like Java's splitByContiguousRowRange, build over physical row IDs.
+    # Query readers apply the deletion vectors from their own snapshot.
     return DataSplit(
         files=list(files),
         partition=split.partition,
         bucket=split.bucket,
         raw_convertible=getattr(split, "raw_convertible", False),
-        data_deletion_files=data_deletion_files,
     )

paimon-python/pypaimon/globalindex/create_global_index.py

@@ -518,7 +518,7 @@ def _validate_unaware_bucket_table(self) -> None:
             )
 
     def _validate_generic_index_table(self) -> None:
-        if (self._core_options.deletion_vectors_enabled()
+        if (self._table.options.deletion_vectors_enabled()
                 and not (self._index_type in VINDEX_IDENTIFIERS
                          and self._table.options.data_evolution_enabled())):
             raise ValueError(

paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py

@@ -25,7 +25,7 @@
     GlobalIndexEvaluator,
 )
 from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
-from pypaimon.globalindex.global_index_reader import GlobalIndexReader, _map_future
+from pypaimon.globalindex.global_index_reader import GlobalIndexReader
 from pypaimon.globalindex.global_index_result import GlobalIndexResult
 from pypaimon.common.options.core_options import CoreOptions
 from pypaimon.common.options.options import Options
@@ -110,6 +110,8 @@ def _create_evaluator(self, fields, file_io, index_path, index_files):
         options = self._options
 
         def readers_function(field: DataField) -> Collection[GlobalIndexReader]:
+            from pypaimon.globalindex.union_global_index_reader import UnionGlobalIndexReader
+
             groups = []
             group = index_metas.get(field.id)
             if group is not None:
@@ -124,25 +126,14 @@ def readers_function(field: DataField) -> Collection[GlobalIndexReader]:
                 )
             if not groups:
                 return []
-            if len(groups) == 1:
-                return _create_readers(
-                    file_io, index_path, groups[0].metas, field, executor, options)
-            union_coverage = Range.sort_and_merge_overlap(
-                [
-                    range_key
-                    for group in groups
-                    for range_key in group.coverage_ranges
-                ],
-                True,
-            )
+            # Like Java, union every reader for the field. Missing coverage
+            # belongs to raw fallback, not to the matches returned by an index.
             readers = []
             for group in groups:
-                pad_ranges = _exclude_ranges(union_coverage, group.coverage_ranges)
                 readers.extend(
                     _create_readers(
-                        file_io, index_path, group.metas, field, executor,
-                        options, pad_ranges=pad_ranges))
-            return readers
+                        file_io, index_path, group.metas, field, executor, options))
+            return [UnionGlobalIndexReader(readers)]
 
         return GlobalIndexEvaluator(fields, readers_function)
 
@@ -272,76 +263,11 @@ def __init__(self, index_field_id, field_ids):
         self.index_field_id = index_field_id
         self.field_ids = tuple(field_ids)
         self.metas = {}
-        self.coverage_ranges = []
 
     def add_file(self, index_type, range_key, io_meta):
-        self.coverage_ranges.append(range_key)
         self.metas.setdefault(index_type, {}).setdefault(range_key, []).append(io_meta)
 
 
-class _PaddingGlobalIndexReader(GlobalIndexReader):
-    def __init__(self, wrapped, padding):
-        self._wrapped = wrapped
-        # Padding rows have not been tested by this index.
-        self._padding = GlobalIndexResult.create(padding.results(), is_exact=False)
-
-    def _pad(self, future):
-        return _map_future(
-            future,
-            lambda result: None if result is None else result.or_(self._padding))
-
-    def visit_equal(self, field_ref, literal):
-        return self._pad(self._wrapped.visit_equal(field_ref, literal))
-
-    def visit_not_equal(self, field_ref, literal):
-        return self._pad(self._wrapped.visit_not_equal(field_ref, literal))
-
-    def visit_less_than(self, field_ref, literal):
-        return self._pad(self._wrapped.visit_less_than(field_ref, literal))
-
-    def visit_less_or_equal(self, field_ref, literal):
-        return self._pad(self._wrapped.visit_less_or_equal(field_ref, literal))
-
-    def visit_greater_than(self, field_ref, literal):
-        return self._pad(self._wrapped.visit_greater_than(field_ref, literal))
-
-    def visit_greater_or_equal(self, field_ref, literal):
-        return self._pad(self._wrapped.visit_greater_or_equal(field_ref, literal))
-
-    def visit_is_null(self, field_ref):
-        return self._pad(self._wrapped.visit_is_null(field_ref))
-
-    def visit_is_not_null(self, field_ref):
-        return self._pad(self._wrapped.visit_is_not_null(field_ref))
-
-    def visit_in(self, field_ref, literals):
-        return self._pad(self._wrapped.visit_in(field_ref, literals))
-
-    def visit_not_in(self, field_ref, literals):
-        return self._pad(self._wrapped.visit_not_in(field_ref, literals))
-
-    def visit_starts_with(self, field_ref, literal):
-        return self._pad(self._wrapped.visit_starts_with(field_ref, literal))
-
-    def visit_ends_with(self, field_ref, literal):
-        return self._pad(self._wrapped.visit_ends_with(field_ref, literal))
-
-    def visit_contains(self, field_ref, literal):
-        return self._pad(self._wrapped.visit_contains(field_ref, literal))
-
-    def visit_like(self, field_ref, literal):
-        return self._pad(self._wrapped.visit_like(field_ref, literal))
-
-    def visit_between(self, field_ref, from_v, to_v):
-        return self._pad(self._wrapped.visit_between(field_ref, from_v, to_v))
-
-    def visit_not_between(self, field_ref, from_v, to_v):
-        return self._pad(self._wrapped.visit_not_between(field_ref, from_v, to_v))
-
-    def close(self):
-        self._wrapped.close()
-
-
 def _resolve_snapshot(table, snapshot):
     if snapshot is not None:
         return snapshot
@@ -392,17 +318,9 @@ def _core_options(table):
     return options
 
 
-def _exclude_ranges(base_ranges, excluded_ranges):
-    excluded_ranges = Range.sort_and_merge_overlap(excluded_ranges, True)
-    result = []
-    for base_range in Range.sort_and_merge_overlap(base_ranges, True):
-        result.extend(base_range.exclude(excluded_ranges))
-    return Range.sort_and_merge_overlap(result, True)
-
-
 def _create_readers(
         file_io, index_path, index_type_metas, field, executor=None,
-        options=None, pad_ranges=None):
+        options=None):
     """Create readers for a specific field, dispatched by index_type.
 
     Unknown indexTypes raise — a silent skip would make
@@ -430,11 +348,7 @@ def _create_readers(
                     OffsetGlobalIndexReader(
                         inner, range_key.from_, range_key.to))
         if offset_readers:
-            reader = UnionGlobalIndexReader(offset_readers)
-            if pad_ranges:
-                padding = GlobalIndexResult.from_ranges(pad_ranges)
-                reader = _PaddingGlobalIndexReader(reader, padding)
-            readers.append(reader)
+            readers.append(UnionGlobalIndexReader(offset_readers))
     return readers
 
 

paimon-python/pypaimon/globalindex/drop_global_index.py

@@ -89,14 +89,18 @@ def __init__(
                 )
 
     def build(self) -> List[CommitMessage]:
+        partition_filter = self._resolve_partition_filter()
+        messages = self._native_build(partition_filter)
+        if messages is not None:
+            return messages
+
         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
@@ -133,6 +137,27 @@ def should_delete(entry: IndexManifestEntry) -> bool:
             for partition, deletes in by_partition.items()
         ]
 
+    def _native_build(self, partition_filter):
+        from pypaimon.read.native_plan import _native_table, _predicate_to_native, native_method_available
+        from pypaimon.write.native_commit import _rest_catalog_supported, from_native_commit_messages
+
+        table = self._table
+        if (not table.options.native_write_enabled()
+                or table.options.query_auth_enabled
+                or not native_method_available('Table', 'new_global_index_drop_builder')
+                or not _rest_catalog_supported(table)):
+            return None
+        from pypaimon_rust.datafusion import GlobalIndexDropBuilder
+        if not GlobalIndexDropBuilder.supports_index_type(self._index_type):
+            return None
+        native = _native_table(table).new_global_index_drop_builder()
+        native.with_index_columns(self._index_columns).with_index_type(self._index_type)
+        if partition_filter is not None:
+            native.with_partition_filter(_predicate_to_native(partition_filter))
+        # Preparing deletion messages is read-only. Python retains explicit
+        # commit and dry-run ownership; failures must not remove index files.
+        return from_native_commit_messages(table, native.build())
+
     def _resolve_partition_filter(self) -> Optional[Predicate]:
         if self._partition_filter is not None:
             return self._partition_filter

paimon-python/pypaimon/globalindex/native_index_build.py

@@ -32,9 +32,7 @@ def build_native_global_index(builder, partition_filter):
     from pypaimon.globalindex.vindex.vindex_vector_global_index_reader import VINDEX_IDENTIFIERS
     if (builder._index_type not in ('btree', 'bitmap', 'full-text') + tuple(VINDEX_IDENTIFIERS)
             or not table.options.native_write_enabled()
-            or not table.options.data_evolution_enabled()
-            or not table.options.global_index_enabled()
-            or table.options.deletion_vectors_enabled()
+            or (table.options.deletion_vectors_enabled() and builder._index_type == 'full-text')
             or table.options.query_auth_enabled
             or table.is_primary_key_table
             or table.options.file_format() != 'parquet'

paimon-python/pypaimon/globalindex/vindex/vindex_vector_global_index_reader.py

@@ -111,6 +111,7 @@ def __init__(self, file_io, index_path, io_metas, options=None):
         self._reader = None
         self._metadata = None
         self._search_params_type = None
+        self._native_reader = False
         self._load_lock = threading.Lock()
 
     def visit_vector_search(self, vector_search):
@@ -134,6 +135,12 @@ def visit_batch_vector_search(self, batch_vector_search):
     def _run_search(self, vectors, limit, include_row_ids, query_options):
         self._ensure_loaded()
 
+        if self._native_reader:
+            results = self._reader.search(
+                vectors, limit, filter_bytes=_filter_bytes(include_row_ids),
+                options={str(key): str(value) for key, value in (query_options or {}).items()})
+            return [DictBasedScoredIndexResult(scores) if scores else None for scores in results]
+
         queries = self._validate_queries(vectors)
         n = len(queries)
         effective_k = self._effective_k(limit, include_row_ids)
@@ -180,6 +187,8 @@ def _validate_queries(self, vectors):
 
     def vector_metric(self):
         self._ensure_loaded()
+        if self._native_reader:
+            return self._reader.metric()
         return self._metadata.metric
 
     def _effective_k(self, limit, include_row_ids):
@@ -200,13 +209,6 @@ def _ensure_loaded(self):
             if self._reader is not None:
                 return
 
-            try:
-                from paimon_vindex import SearchParams, VectorIndexReader
-            except ImportError as e:
-                raise ImportError(
-                    "paimon-vindex is required to read vindex vector indexes. "
-                    "Install paimon-vindex==0.5.0 or pypaimon[vindex].") from e
-
             file_path = (self._io_meta.external_path
                          if self._io_meta.external_path
                          else os.path.join(self._index_path, self._io_meta.file_name))
@@ -216,12 +218,26 @@ def _ensure_loaded(self):
             index_input = None
             try:
                 index_input = PaimonVindexInput(stream, parallelism)
-                reader = VectorIndexReader(index_input)
-                self._metadata = reader.metadata()
-                reader.optimize_for_search()
+                native_reader = _native_reader_type()
+                if native_reader is not None:
+                    reader = native_reader(
+                        index_input, file_path, self._io_meta.file_size,
+                        {str(key): str(value) for key, value in self._options.items()})
+                    self._native_reader = True
+                else:
+                    try:
+                        from paimon_vindex import SearchParams, VectorIndexReader
+                    except ImportError as e:
+                        raise ImportError(
+                            "paimon-vindex is required to read vindex vector indexes. "
+                            "Install paimon-vindex==0.5.0 or pypaimon[vindex].") from e
+                    reader = VectorIndexReader(index_input)
+                    self._metadata = reader.metadata()
+                    reader.optimize_for_search()
+                    self._search_params_type = SearchParams
+                    self._native_reader = False
                 self._index_input = index_input
                 self._reader = reader
-                self._search_params_type = SearchParams
                 self._stream = stream
             except BaseException:
                 try:
@@ -274,11 +290,17 @@ def _read_parallelism(options, file_path):
     return parallelism
 
 
+def _native_reader_type():
+    try:
+        from pypaimon_rust import datafusion
+    except ImportError:
+        return None
+    return getattr(datafusion, 'VindexVectorGlobalIndexReader', None)
+
+
 def _filter_bytes(include_row_ids):
     if include_row_ids is None:
         return None
-    if include_row_ids.cardinality() == 0:
-        return None
     return include_row_ids.serialize()
 
 
@@ -359,7 +381,14 @@ def _int_parameter(options, key):
     if value is None:
         return None
     try:
-        return int(value)
+        text = str(value)
+        digits = text[1:] if text[:1] in ('+', '-') else text
+        if not digits or any(ord(char) > 0xffff or not char.isdecimal() for char in digits):
+            raise ValueError('Invalid Java integer text')
+        number = int(text)
+        if not -(1 << 31) <= number < (1 << 31):
+            raise ValueError('Java integer overflow')
+        return number
     except (TypeError, ValueError) as e:
         raise ValueError(
             "Invalid value for '%s': %s. Must be an integer." % (key, value)) from e

paimon-python/pypaimon/ray/batch_vector_search.py

@@ -31,7 +31,8 @@
 
 def _execute_batch_vector_search(builder, *, concurrency=None, ray_remote_args=None):
     concurrency, remote_args = _execution_options(concurrency, ray_remote_args)
-    reader = builder.new_batch_vector_search_read()
+    from pypaimon.table.source.native_search_plan import python_search_reader
+    reader = python_search_reader(builder.new_batch_vector_search_read())
     if (type(reader) is not BatchVectorSearchReadImpl
             or not reader._table.options.data_evolution_enabled()):
         raise ValueError("Ray vector search supports only data-evolution tables.")

paimon-python/pypaimon/ray/full_text_search.py

@@ -28,7 +28,8 @@
 
 def _execute_full_text_search(builder, *, concurrency=None, ray_remote_args=None):
     concurrency, remote_args = _execution_options(concurrency, ray_remote_args)
-    reader = builder.new_full_text_read()
+    from pypaimon.table.source.native_search_plan import python_search_reader
+    reader = python_search_reader(builder.new_full_text_read())
     if (type(reader) is not DataEvolutionFullTextRead
             or not reader._table.options.data_evolution_enabled()):
         raise ValueError("Ray full-text search supports only data-evolution tables.")

paimon-python/pypaimon/ray/vector_search.py

@@ -38,7 +38,8 @@
 def _execute_vector_search(builder, *, concurrency=None, ray_remote_args=None):
     """Execute the builder of an already snapshot-pinned ``VectorQuery``."""
     concurrency, remote_args = _execution_options(concurrency, ray_remote_args)
-    reader = builder.new_vector_search_read()
+    from pypaimon.table.source.native_search_plan import python_search_reader
+    reader = python_search_reader(builder.new_vector_search_read())
     if (type(reader) is not DataEvolutionVectorRead
             or not reader._table.options.data_evolution_enabled()):
         raise ValueError("Ray vector search supports only data-evolution tables.")