diff --git a/docs/docs/multimodal-table/global-index/manage-indexes.mdx b/docs/docs/multimodal-table/global-index/manage-indexes.mdx index 2e1ff0c9f9cf..00c8bb94c40d 100644 --- a/docs/docs/multimodal-table/global-index/manage-indexes.mdx +++ b/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 diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 7b0665c50296..c2a4f84929ac 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/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. " diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java index 47aa94acd942..2855235a236f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java +++ b/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 diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 903697a94ff2..84050cef0d2e 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/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" diff --git a/paimon-python/pypaimon/multimodal/query.py b/paimon-python/pypaimon/multimodal/query.py index 489ff641525b..fca13b932592 100644 --- a/paimon-python/pypaimon/multimodal/query.py +++ b/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, diff --git a/paimon-python/pypaimon/read/read_builder.py b/paimon-python/pypaimon/read/read_builder.py index a531d5d2c83f..33892608f2d6 100644 --- a/paimon-python/pypaimon/read/read_builder.py +++ b/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`` diff --git a/paimon-python/pypaimon/read/table_scan.py b/paimon-python/pypaimon/read/table_scan.py index 884d33b3cb5e..5e5cffa8fd60 100755 --- a/paimon-python/pypaimon/read/table_scan.py +++ b/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 diff --git a/paimon-python/pypaimon/tests/global_index_build_test.py b/paimon-python/pypaimon/tests/global_index_build_test.py index 186270fcb79a..cd44dbfce3a4 100644 --- a/paimon-python/pypaimon/tests/global_index_build_test.py +++ b/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( diff --git a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py index 84d3e9468380..f141aa4b0b75 100644 --- a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py +++ b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py @@ -90,6 +90,15 @@ def test_family_modes_override_global_mode(self): self.assertEqual( GlobalIndexSearchMode.FAST, options.full_text_index_search_mode()) + def test_adaptive_scalar_mode(self): + options = CoreOptions(Options({ + "scalar-index.search-mode": "adaptive", + })) + self.assertEqual( + GlobalIndexSearchMode.ADAPTIVE, + options.scalar_index_search_mode(), + ) + def test_coverage_honours_search_mode_override(self): coverage = _coverage(CoreOptions(Options.from_none())) self.assertEqual(