You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/04/29 14:06:22 UTC
(paimon) branch master updated: [python] Push down IndexedSplit row ranges to Lance reader (#7666)
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 abf1ca6559 [python] Push down IndexedSplit row ranges to Lance reader (#7666)
abf1ca6559 is described below
commit abf1ca65594d81ef36643b52fbc7fd8ecdc6323b
Author: ChengHui Chen <ch...@gmail.com>
AuthorDate: Wed Apr 29 22:06:15 2026 +0800
[python] Push down IndexedSplit row ranges to Lance reader (#7666)
After a global index search returns a sparse set of matching row IDs,
the data fetch path previously used RowIdFilterRecordBatchReader for
Lance — reading the entire Lance file and discarding non-matching rows
in memory.
This change converts the global row ID ranges to local file offsets and
pushes them down to lance.file.LanceFileReader.take_rows(), so only the
matched rows are physically read from disk, leveraging Lance's native
random-access capability.
---
.../pypaimon/read/reader/format_lance_reader.py | 15 +-
paimon-python/pypaimon/read/split_read.py | 19 ++-
.../pypaimon/tests/reader_append_only_test.py | 166 +++++++++++++++++++++
3 files changed, 193 insertions(+), 7 deletions(-)
diff --git a/paimon-python/pypaimon/read/reader/format_lance_reader.py b/paimon-python/pypaimon/read/reader/format_lance_reader.py
index da68a994ab..8427560f0c 100644
--- a/paimon-python/pypaimon/read/reader/format_lance_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_lance_reader.py
@@ -16,7 +16,7 @@
# limitations under the License.
################################################################################
-from typing import List, Optional, Any
+from typing import Any, List, Optional, Tuple
import pyarrow as pa
import pyarrow.dataset as ds
@@ -36,7 +36,9 @@ class FormatLanceReader(RecordBatchReader):
"""
def __init__(self, file_io: FileIO, file_path: str, read_fields: List[DataField],
- push_down_predicate: Any, batch_size: int = 1024):
+ push_down_predicate: Any, batch_size: int = 1024,
+ row_range: Optional[Tuple[int, int]] = None,
+ row_indices: Optional[List[int]] = None):
"""Initialize Lance reader."""
import lance
@@ -61,7 +63,14 @@ class FormatLanceReader(RecordBatchReader):
file_path_for_lance,
storage_options=storage_options,
columns=columns_for_lance)
- pa_table = lance_reader.read_all().to_table()
+ if row_indices is not None:
+ reader_results = lance_reader.take_rows(row_indices)
+ elif row_range is not None:
+ start, end = row_range
+ reader_results = lance_reader.read_range(start, end - start)
+ else:
+ reader_results = lance_reader.read_all()
+ pa_table = reader_results.to_table()
# Precompute output schema for missing fields
if self.missing_fields:
diff --git a/paimon-python/pypaimon/read/split_read.py b/paimon-python/pypaimon/read/split_read.py
index 842d77095f..9274377f4a 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -130,7 +130,8 @@ class SplitRead(ABC):
def file_reader_supplier(self, file: DataFileMeta, for_merge_read: bool,
read_fields: List[str], row_tracking_enabled: bool,
- row_ranges=None) -> RecordBatchReader:
+ row_ranges=None,
+ row_range: Optional[Tuple[int, int]] = None) -> RecordBatchReader:
(read_file_fields, read_arrow_predicate) = self._get_fields_and_predicate(file.schema_id, read_fields)
# Use external_path if available, otherwise use file_path
@@ -145,8 +146,8 @@ class SplitRead(ABC):
effective_row_ranges = Range.and_(row_ranges, [file.row_id_range()])
if len(effective_row_ranges) == 0:
return EmptyRecordBatchReader()
- if file_format == CoreOptions.FILE_FORMAT_VORTEX:
- # Convert global row ranges to local indices for Vortex pushdown
+ if file_format in (CoreOptions.FILE_FORMAT_VORTEX, CoreOptions.FILE_FORMAT_LANCE):
+ # Convert global row ranges to local file indices for native pushdown
row_indices = []
for r in effective_row_ranges:
start = r.from_ - file.first_row_id
@@ -166,7 +167,8 @@ class SplitRead(ABC):
name_to_field = {f.name: f for f in self.read_fields}
ordered_read_fields = [name_to_field[n] for n in read_file_fields if n in name_to_field]
format_reader = FormatLanceReader(self.table.file_io, file_path, ordered_read_fields,
- read_arrow_predicate, batch_size=batch_size)
+ read_arrow_predicate, batch_size=batch_size,
+ row_range=row_range, row_indices=row_indices)
elif file_format == CoreOptions.FILE_FORMAT_VORTEX:
name_to_field = {f.name: f for f in self.read_fields}
ordered_read_fields = [name_to_field[n] for n in read_file_fields if n in name_to_field]
@@ -419,6 +421,15 @@ class RawFileSplitRead(SplitRead):
(start_pos, end_pos) = shard_file_idx_map[file.file_name]
if (start_pos, end_pos) == (-1, -1):
return None
+ file_path = file.external_path if file.external_path else file.file_path
+ file_format = format_identifier(os.path.basename(file_path))
+ if file_format == CoreOptions.FILE_FORMAT_LANCE:
+ file_batch_reader = self.file_reader_supplier(
+ file=file,
+ for_merge_read=False,
+ read_fields=read_fields,
+ row_tracking_enabled=True,
+ row_range=(start_pos, end_pos))
else:
file_batch_reader = ShardBatchReader(self.file_reader_supplier(
file=file,
diff --git a/paimon-python/pypaimon/tests/reader_append_only_test.py b/paimon-python/pypaimon/tests/reader_append_only_test.py
index c4763dd99e..00b2f803f7 100644
--- a/paimon-python/pypaimon/tests/reader_append_only_test.py
+++ b/paimon-python/pypaimon/tests/reader_append_only_test.py
@@ -248,6 +248,172 @@ class AoReaderTest(unittest.TestCase):
])
self.assertEqual(actual.sort_by('user_id'), expected)
+ def test_lance_sliced_split_row_range_pushdown(self):
+ """
+ SlicedSplit with Lance format calls read_range() instead of read_all(),
+ reading only the requested row slice from disk rather than the full file.
+ """
+ import unittest.mock as mock
+ try:
+ import lance as _lance
+ import lance.file # ensure submodule is loaded before write_lance uses it
+ except ImportError:
+ self.skipTest("lance not installed")
+
+ schema = Schema.from_pyarrow_schema(
+ pa.schema([('id', pa.int64()), ('value', pa.string())]),
+ options={'file.format': 'lance'})
+ self.catalog.create_table('default.test_lance_sliced_split', schema, False)
+ table = self.catalog.get_table('default.test_lance_sliced_split')
+
+ # Write 1000 rows in a single shot so they land in one file
+ n = 1000
+ pa_table = pa.table({'id': list(range(n)), 'value': [f'v{i}' for i in range(n)]})
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_write.write_arrow(pa_table)
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+
+ scan = table.new_read_builder().new_scan()
+ splits = scan.plan().splits()
+ self.assertEqual(len(splits), 1, "Expected single split (no partition, single bucket)")
+ data_split = splits[0]
+ self.assertEqual(len(data_split.files), 1, "Expected single file in split")
+
+ file_name = data_split.files[0].file_name
+ slice_start, slice_end = 200, 700 # request 500 of 1000 rows
+
+ from pypaimon.read.sliced_split import SlicedSplit
+ sliced = SlicedSplit(data_split, {file_name: (slice_start, slice_end)})
+
+ # Spy on lance.file.LanceFileReader to verify read_range is used
+ OrigReader = _lance.file.LanceFileReader
+ read_all_calls = []
+ read_range_calls = []
+
+ class SpyLanceFileReader:
+ def __init__(self, *args, **kwargs):
+ self._r = OrigReader(*args, **kwargs)
+
+ def read_all(self, *args, **kwargs):
+ read_all_calls.append(())
+ return self._r.read_all(*args, **kwargs)
+
+ def read_range(self, start, count, **kwargs):
+ read_range_calls.append((start, count))
+ return self._r.read_range(start, count, **kwargs)
+
+ def __getattr__(self, name):
+ return getattr(self._r, name)
+
+ table_read = table.new_read_builder().new_read()
+ with mock.patch.object(_lance.file, 'LanceFileReader', SpyLanceFileReader):
+ result = table_read.to_arrow([sliced])
+
+ # Verify that read_range was used, not read_all
+ self.assertEqual(len(read_all_calls), 0,
+ "read_all() must not be called when SlicedSplit row range is available")
+ self.assertEqual(len(read_range_calls), 1)
+ actual_start, actual_count = read_range_calls[0]
+ self.assertEqual(actual_start, slice_start)
+ self.assertEqual(actual_count, slice_end - slice_start,
+ f"read_range should request exactly {slice_end - slice_start} rows, "
+ f"not the full {n} rows of the file")
+
+ # Verify functional correctness: correct rows returned
+ self.assertEqual(result.num_rows, slice_end - slice_start)
+ result_ids = sorted(result.column('id').to_pylist())
+ self.assertEqual(result_ids, list(range(slice_start, slice_end)))
+
+ def test_lance_indexed_split_take_rows_pushdown(self):
+ """
+ IndexedSplit row ranges (from ANN global index results) are converted to
+ local file indices and pushed down to lance.file.LanceFileReader.take_rows(),
+ so only the matched rows are physically read instead of the full file.
+ """
+ import unittest.mock as mock
+ try:
+ import lance as _lance
+ import lance.file
+ except ImportError:
+ self.skipTest("lance not installed")
+
+ schema = Schema.from_pyarrow_schema(
+ pa.schema([('id', pa.int64()), ('value', pa.string())]),
+ options={'file.format': 'lance',
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true'})
+ self.catalog.create_table('default.test_lance_indexed_split', schema, False)
+ table = self.catalog.get_table('default.test_lance_indexed_split')
+
+ n = 1000
+ pa_table = pa.table({'id': list(range(n)), 'value': [f'v{i}' for i in range(n)]})
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_write.write_arrow(pa_table)
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+
+ splits = table.new_read_builder().new_scan().plan().splits()
+ self.assertEqual(len(splits), 1)
+ data_split = splits[0]
+ self.assertEqual(len(data_split.files), 1)
+
+ file_meta = data_split.files[0]
+ self.assertIsNotNone(file_meta.first_row_id, "row tracking must assign first_row_id")
+ first_row_id = file_meta.first_row_id
+
+ # Simulate ANN result: 3 scattered global row IDs
+ from pypaimon.globalindex.indexed_split import IndexedSplit
+ from pypaimon.utils.range import Range
+ target_global_ids = [first_row_id + 50, first_row_id + 300, first_row_id + 700]
+ row_ranges = [Range(g, g) for g in target_global_ids]
+ indexed = IndexedSplit(data_split, row_ranges)
+
+ OrigReader = _lance.file.LanceFileReader
+ take_rows_calls = []
+ read_all_calls = []
+
+ class SpyLanceFileReader:
+ def __init__(self, *args, **kwargs):
+ self._r = OrigReader(*args, **kwargs)
+
+ def take_rows(self, indices, **kwargs):
+ take_rows_calls.append(list(indices))
+ return self._r.take_rows(indices, **kwargs)
+
+ def read_all(self, *args, **kwargs):
+ read_all_calls.append(())
+ return self._r.read_all(*args, **kwargs)
+
+ def read_range(self, *args, **kwargs):
+ return self._r.read_range(*args, **kwargs)
+
+ def __getattr__(self, name):
+ return getattr(self._r, name)
+
+ table_read = table.new_read_builder().new_read()
+ with mock.patch.object(_lance.file, 'LanceFileReader', SpyLanceFileReader):
+ result = table_read.to_arrow([indexed])
+
+ self.assertEqual(len(read_all_calls), 0,
+ "read_all() must not be called when IndexedSplit row ranges are available")
+ self.assertEqual(len(take_rows_calls), 1)
+ # local indices = global_ids - first_row_id
+ expected_local = [g - first_row_id for g in target_global_ids]
+ self.assertEqual(sorted(take_rows_calls[0]), sorted(expected_local),
+ f"take_rows should request local indices {expected_local}, "
+ f"not read the full {n} rows")
+
+ self.assertEqual(result.num_rows, len(target_global_ids))
+ result_ids = sorted(result.column('id').to_pylist())
+ self.assertEqual(result_ids, [50, 300, 700])
+
def test_append_only_multi_write_once_commit(self):
schema = Schema.from_pyarrow_schema(self.pa_schema, partition_keys=['dt'])
self.catalog.create_table('default.test_append_only_multi_once_commit', schema, False)