[python] Add adaptive scalar index reads for LIMIT queries

docs/docs/multimodal-table/global-index/manage-indexes.mdx

@@ -235,6 +235,7 @@ Updates to already indexed rows also require the [update policy](#update-indexed
 | --- | --- | --- |
 | `fast` | Search indexed coverage only. | The index is current, or partial coverage is acceptable. |
 | `full` | Check row-ID coverage against snapshot `nextRowId`; scan raw data for detected gaps where supported. | Newly appended ranges need to be included. |
+| `adaptive` | For a limited PyPaimon Arrow read, search indexed rows first and scan uncovered ranges only if the limit is not reached. Other reads use `full` semantics. | Point or small-result scalar lookups that must include newly appended rows. |
 | `detail` | Compare active data-file row-ID ranges with index coverage, then scan detected gaps where supported. | Coverage should be checked against current files, including partition filtering. |
 
 ![An index covers the original row range; appended rows need another index build or a search mode that includes uncovered data.](/img/multimodal-index-coverage.svg)
@@ -279,6 +280,12 @@ the latest column values. An update that preserves row IDs can therefore leave
 stale index entries even when coverage is complete. Handle those updates through
 the [index update policy](#update-indexed-columns) and an index build.
 
+PyPaimon's `adaptive` scalar mode optimizes materialized `ReadBuilder.to_arrow()`
+and multimodal `scan(...).to_arrow()` / `read_blobs()` calls with a finite limit.
+Both stages use the same snapshot, and fallback is skipped only after all read
+filters have run. Without a limit, or through a static `new_scan().plan()`, it
+behaves like `full`.
+
 Use `scalar-index.search-mode`, `vector-index.search-mode`, or
 `full-text-index.search-mode` for one index family. The legacy
 `global-index.search-mode` has no default and is used as a fallback when the corresponding

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -5965,6 +5965,11 @@ public enum GlobalIndexSearchMode implements DescribedEnum {
                 "full",
                 "Use snapshot next row id and global index coverage to detect missing row ids, "
                         + "and scan raw data only when a gap exists."),
+        ADAPTIVE(
+                "adaptive",
+                "For supported limited PyPaimon reads, search indexed data first and scan "
+                        + "unindexed rows only when the limit is not reached. Other reads use "
+                        + "full fallback semantics."),
         DETAIL(
                 "detail",
                 "Scan data files to find exact unindexed rows. "

paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java

@@ -159,6 +159,11 @@ public void testIndexSearchModes() {
                 .isEqualTo(CoreOptions.GlobalIndexSearchMode.FULL);
         assertThat(options.fullTextIndexSearchMode())
                 .isEqualTo(CoreOptions.GlobalIndexSearchMode.FULL);
+
+        conf.setString(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), "adaptive");
+        options = new CoreOptions(conf);
+        assertThat(options.scalarIndexSearchMode())
+                .isEqualTo(CoreOptions.GlobalIndexSearchMode.ADAPTIVE);
     }
 
     @Test

paimon-python/pypaimon/common/options/core_options.py

@@ -96,6 +96,7 @@ class GlobalIndexColumnUpdateAction(str, Enum):
 class GlobalIndexSearchMode(str, Enum):
     FAST = "fast"
     FULL = "full"
+    ADAPTIVE = "adaptive"
     DETAIL = "detail"
 
 

paimon-python/pypaimon/multimodal/query.py

@@ -68,9 +68,7 @@ def to_arrow(self):
             return self._read_global_index_result(self._result_factory(self))
 
         read_builder = self._configured_read_builder()
-        scan = read_builder.new_scan()
-        plan = scan.plan()
-        return read_builder.new_read().to_arrow(plan.splits())
+        return read_builder.to_arrow()
 
     def to_arrow_batch_reader(self, *, blob_parallelism=None):
         """Stream this scan as Arrow batches without collecting a table."""
@@ -273,8 +271,7 @@ def read_blobs(
         """
         blob_cols = self._resolve_blob_columns(columns)
         read_builder, file_io = self._blob_descriptor_read_builder(blob_cols)
-        arrow = read_builder.new_read().to_arrow(
-            read_builder.new_scan().plan().splits())
+        arrow = read_builder.to_arrow()
         map_blob_cols = set(blob_cols) - set(self._all_blob_columns())
         bodies = self._fetch_bodies(
             file_io, arrow.select(blob_cols).to_pydict(), blob_cols,

paimon-python/pypaimon/read/read_builder.py

@@ -17,6 +17,12 @@
 
 from typing import List, Optional
 
+import pyarrow as pa
+
+from pypaimon.common.options.core_options import (
+    CoreOptions,
+    GlobalIndexSearchMode,
+)
 from pypaimon.common.predicate import Predicate
 from pypaimon.common.predicate_builder import PredicateBuilder
 from pypaimon.read.explain import ExplainResult, ExplainSplitInfo, PruningStat
@@ -98,6 +104,137 @@ def new_read(self) -> TableRead:
             limit=self._limit,
         )
 
+    def to_arrow(
+        self,
+        parallelism: Optional[int] = None,
+        blob_parallelism: Optional[int] = None,
+    ) -> pa.Table:
+        """Plan and materialize this read as an Arrow table.
+
+        In adaptive scalar-index mode, supported limited reads first read the
+        indexed ranges and plan uncovered ranges only when the filtered result
+        does not reach the limit. Other cases retain FULL semantics.
+        """
+        if self._adaptive_scalar_read_supported():
+            result = self._adaptive_scalar_to_arrow(
+                parallelism, blob_parallelism)
+            if result is not None:
+                return result
+        return self._read_once(parallelism, blob_parallelism)
+
+    def _read_once(self, parallelism, blob_parallelism) -> pa.Table:
+        plan = self.new_scan().plan()
+        return self.new_read().to_arrow(
+            plan.splits(),
+            parallelism=parallelism,
+            blob_parallelism=blob_parallelism,
+        )
+
+    def _adaptive_scalar_read_supported(self) -> bool:
+        options = self.table.options
+        return (
+            options.scalar_index_search_mode()
+            == GlobalIndexSearchMode.ADAPTIVE
+            and self._limit is not None
+            and self._limit > 0
+            and self._predicate is not None
+            and options.data_evolution_enabled()
+            and not self.table.is_primary_key_table
+            and not options.options.contains(
+                CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP)
+        )
+
+    def _adaptive_scalar_to_arrow(
+        self,
+        parallelism: Optional[int],
+        blob_parallelism: Optional[int],
+    ) -> Optional[pa.Table]:
+        snapshot = self._target_snapshot()
+        if snapshot is None:
+            return None
+
+        pinned = self._table_at_snapshot(
+            snapshot.id, GlobalIndexSearchMode.ADAPTIVE)
+        evaluation_builder = self._copy_for_table(pinned)
+        index_plan = (
+            evaluation_builder.new_scan().file_scanner._eval_global_index(
+                snapshot)
+        )
+
+        from pypaimon.read.scanner.file_scanner import (
+            _GlobalIndexPlanningResult,
+        )
+        if not isinstance(index_plan, _GlobalIndexPlanningResult):
+            return evaluation_builder._read_once(
+                parallelism, blob_parallelism)
+
+        indexed_table = self._table_at_snapshot(
+            snapshot.id, GlobalIndexSearchMode.FAST)
+        indexed_builder = self._copy_for_table(indexed_table)
+        indexed_scan = indexed_builder.new_scan().with_global_index_result(
+            index_plan.indexed_result)
+        indexed = indexed_builder.new_read().to_arrow(
+            indexed_scan.plan().splits(),
+            parallelism=parallelism,
+            blob_parallelism=blob_parallelism,
+        )
+        if indexed.num_rows >= self._limit or not index_plan.unindexed_ranges:
+            return indexed
+
+        fallback_table = self._table_at_snapshot(
+            snapshot.id, GlobalIndexSearchMode.FULL)
+        fallback_builder = self._copy_for_table(
+            fallback_table, limit=self._limit - indexed.num_rows)
+        fallback_scan = fallback_builder.new_scan().with_row_ranges(
+            index_plan.unindexed_ranges)
+        fallback = fallback_builder.new_read().to_arrow(
+            fallback_scan.plan().splits(),
+            parallelism=parallelism,
+            blob_parallelism=blob_parallelism,
+        )
+        if indexed.num_rows == 0:
+            return fallback
+        if fallback.num_rows == 0:
+            return indexed
+        return pa.concat_tables([indexed, fallback])
+
+    def _target_snapshot(self):
+        from pypaimon.snapshot.time_travel_util import TimeTravelUtil
+
+        manager = self.table.snapshot_manager()
+        snapshot = TimeTravelUtil.try_travel_to_snapshot(
+            self.table.options.options,
+            self.table.tag_manager(),
+            manager,
+        )
+        return (
+            snapshot if snapshot is not None else manager.get_latest_snapshot()
+        )
+
+    def _table_at_snapshot(self, snapshot_id, search_mode):
+        from pypaimon.snapshot.time_travel_util import SCAN_KEYS
+
+        options = self.table.options.options
+        overrides = {
+            key: None for key in SCAN_KEYS if options.contains_key(key)
+        }
+        overrides.update({
+            CoreOptions.SCAN_SNAPSHOT_ID.key(): str(snapshot_id),
+            CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(): search_mode.value,
+        })
+        return self.table.copy(overrides)
+
+    def _copy_for_table(self, table, limit=None):
+        builder = ReadBuilder(table)
+        if self._predicate is not None:
+            builder.with_filter(self._predicate)
+        if self._partition_filter is not None:
+            builder.with_partition_filter(self._partition_filter)
+        if self._projection is not None:
+            builder.with_projection(self._projection)
+        builder.with_limit(self._limit if limit is None else limit)
+        return builder
+
     def _nested_name_paths(self) -> Optional[List[List[str]]]:
         """Resolve the current nested-projection state into a parallel list
         of name paths against the underlying table schema. Returns ``None``

paimon-python/pypaimon/read/table_scan.py

@@ -21,7 +21,10 @@
 
 from pypaimon.catalog.catalog_exception import TableNoPermissionException
 from pypaimon.common.identifier import UNKNOWN_DATABASE
-from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.core_options import (
+    CoreOptions,
+    GlobalIndexSearchMode,
+)
 from pypaimon.common.predicate import Predicate
 from pypaimon.common.predicate_builder import PredicateBuilder
 from pypaimon.manifest.manifest_list_manager import ManifestListManager
@@ -164,6 +167,12 @@ def _native_plan_supported_impl(self) -> bool:
         if self.table.bucket_mode() in (BucketMode.HASH_DYNAMIC, BucketMode.CROSS_PARTITION):
             return False
         options = self.table.options.options
+        # A static plan cannot decide whether indexed rows satisfy LIMIT after
+        # residual filtering. ReadBuilder.to_arrow handles that two-stage path;
+        # other adaptive reads retain Python FULL fallback semantics.
+        if (self.table.options.scalar_index_search_mode()
+                == GlobalIndexSearchMode.ADAPTIVE):
+            return False
         if (any(options.contains_key(key)
                 for key in _NATIVE_FAMILY_SEARCH_MODE_OPTIONS)):
             from pypaimon.read.native_plan import native_family_search_modes_available

paimon-python/pypaimon/tests/global_index_build_test.py

@@ -16,6 +16,7 @@
 # under the License.
 
 import unittest
+from unittest.mock import patch
 from datetime import date, datetime
 from decimal import Decimal
 import os
@@ -25,6 +26,8 @@
 
 import pyarrow as pa
 
+from pypaimon.common.options.core_options import GlobalIndexSearchMode
+from pypaimon.common.predicate_builder import PredicateBuilder
 from pypaimon.globalindex.build_plan import (
     filter_non_indexable_splits as _filter_non_indexable_splits,
     split_by_global_index_shard as _split_by_global_index_shard,
@@ -199,6 +202,120 @@ class GlobalIndexBuildTest(
         'file.format': 'parquet',
     }
 
+    def _adaptive_builder(self, table, predicate, limit=None):
+        adaptive = table.copy({'scalar-index.search-mode': 'adaptive'})
+        builder = adaptive.new_read_builder().with_filter(predicate)
+        if limit is not None:
+            builder.with_limit(limit)
+        return builder
+
+    def test_adaptive_scalar_index_reads_fallback_only_when_needed(self):
+        table = self._create_table()
+        self._write_arrow(table, pa.table({
+            'id': [1], 'name': ['indexed'], 'age': [10], 'city': ['old'],
+        }, schema=self.pa_schema))
+        table.create_global_index('id')
+        self._write_arrow(table, pa.table({
+            'id': [1, 2],
+            'name': ['unindexed-residual', 'unindexed-miss'],
+            'age': [20, 30],
+            'city': ['new', 'raw'],
+        }, schema=self.pa_schema))
+
+        pb = table.new_read_builder().new_predicate_builder()
+        indexed_builder = self._adaptive_builder(
+            table,
+            PredicateBuilder.and_predicates([
+                pb.equal('id', 1), pb.equal('city', 'old')]),
+            limit=1,
+        )
+        from pypaimon.read.table_scan import TableScan
+        with patch.object(
+                TableScan,
+                'with_row_ranges',
+                side_effect=AssertionError('raw fallback was planned')):
+            indexed = indexed_builder.to_arrow()
+        self.assertEqual(['indexed'], indexed.column('name').to_pylist())
+
+        raw = self._adaptive_builder(
+            table, pb.equal('id', 2), limit=1).to_arrow()
+        self.assertEqual(['unindexed-miss'], raw.column('name').to_pylist())
+
+        residual = self._adaptive_builder(
+            table,
+            PredicateBuilder.and_predicates([
+                pb.equal('id', 1), pb.equal('city', 'new')]),
+            limit=1,
+        ).to_arrow()
+        self.assertEqual(
+            ['unindexed-residual'], residual.column('name').to_pylist())
+
+        partial = self._adaptive_builder(
+            table, pb.equal('id', 1), limit=2).to_arrow()
+        self.assertEqual(
+            {'indexed', 'unindexed-residual'},
+            set(partial.column('name').to_pylist()),
+        )
+        self.assertEqual(2, partial.num_rows)
+
+    def test_adaptive_scalar_index_pins_fallback_snapshot(self):
+        table = self._create_table()
+        self._write_arrow(table, pa.table({
+            'id': [1], 'name': ['indexed'], 'age': [10], 'city': ['old'],
+        }, schema=self.pa_schema))
+        table.create_global_index('id')
+        self._write_arrow(table, pa.table({
+            'id': [2], 'name': ['visible'], 'age': [20], 'city': ['raw'],
+        }, schema=self.pa_schema))
+
+        pb = table.new_read_builder().new_predicate_builder()
+        builder = self._adaptive_builder(table, pb.equal('id', 2), limit=2)
+        from pypaimon.read.table_read import TableRead
+        original_to_arrow = TableRead.to_arrow
+        appended = [False]
+
+        def append_after_indexed_read(table_read, *args, **kwargs):
+            result = original_to_arrow(table_read, *args, **kwargs)
+            if (not appended[0]
+                    and table_read.table.options.scalar_index_search_mode()
+                    == GlobalIndexSearchMode.FAST):
+                appended[0] = True
+                self._write_arrow(table, pa.table({
+                    'id': [2], 'name': ['too-new'],
+                    'age': [30], 'city': ['raw'],
+                }, schema=self.pa_schema))
+            return result
+
+        with patch.object(
+                TableRead, 'to_arrow', new=append_after_indexed_read):
+            result = builder.to_arrow()
+
+        self.assertTrue(appended[0])
+        self.assertEqual(['visible'], result.column('name').to_pylist())
+
+    def test_adaptive_without_limit_keeps_full_semantics(self):
+        table = self._create_table()
+        self._write_arrow(table, pa.table({
+            'id': [1], 'name': ['indexed'], 'age': [10], 'city': ['old'],
+        }, schema=self.pa_schema))
+        table.create_global_index('id')
+        self._write_arrow(table, pa.table({
+            'id': [2], 'name': ['unindexed'], 'age': [20], 'city': ['new'],
+        }, schema=self.pa_schema))
+
+        pb = table.new_read_builder().new_predicate_builder()
+        builder = self._adaptive_builder(table, pb.equal('id', 2))
+        result = builder.to_arrow()
+
+        self.assertEqual(['unindexed'], result.column('name').to_pylist())
+
+        limited = self._adaptive_builder(
+            table, pb.equal('id', 2), limit=1)
+        plan = limited.new_scan().plan()
+        result = limited.new_read().to_arrow(plan.splits())
+
+        self.assertEqual(['unindexed'], result.column('name').to_pylist())
+
     def test_create_btree_global_index_from_python(self):
         table = self._create_table()
         self._write_arrow(table, pa.table(