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 f6d0574b11 [python] Support positional reads for small Jindo ranges
(#10817)
f6d0574b11 is described below
commit f6d0574b11123b4f7abd89c9688a2123dd6868b9
Author: Jiajia Li <[email protected]>
AuthorDate: Tue Oct 27 14:15:38 2026 +0800
[python] Support positional reads for small Jindo ranges (#10817)
---
paimon-python/pypaimon/common/options/config.py | 8 +
.../filesystem/jindo_file_system_handler.py | 87 ++++++++--
.../pypaimon/tests/jindo_file_system_test.py | 185 +++++++++++++++++++++
3 files changed, 270 insertions(+), 10 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/config.py
b/paimon-python/pypaimon/common/options/config.py
index bac72d4a4c..0d38bd58ef 100644
--- a/paimon-python/pypaimon/common/options/config.py
+++ b/paimon-python/pypaimon/common/options/config.py
@@ -24,6 +24,14 @@ class OssOptions:
"OSS filesystem implementation: legacy, jindo, or cpp (native reads
only)")
OSS_PYTHON_IMPL =
ConfigOptions.key("fs.oss.python.impl").string_type().default_value("jindo").with_description(
"Python OSS implementation when fs.oss.impl=cpp: jindo or legacy")
+ OSS_JINDO_PREAD_MAX_SIZE = (
+ ConfigOptions.key("fs.oss.jindo.pread-max-size")
+ .memory_type()
+ .default_value(MemorySize.of_bytes(0))
+ .with_description(
+ "Max size of a Jindo read after a seek that uses pread instead of
the "
+ "read-ahead stream. 0 disables it.")
+ )
OSS_ACCESS_KEY_ID =
ConfigOptions.key("fs.oss.accessKeyId").string_type().no_default_value().with_description(
"OSS access key ID")
OSS_ACCESS_KEY_SECRET = ConfigOptions.key(
diff --git a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
index 93d255f4a0..25db0ba1a7 100644
--- a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
+++ b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
@@ -49,7 +49,11 @@ from pypaimon.filesystem import oss_user_agent
_JINDO_CONFIG_PREFIXES = ("fs.", "logger.")
_PYPAIMON_ONLY_JINDO_CONFIG_KEYS = {
- OssOptions.OSS_IMPL.key(), OssOptions.OSS_PYTHON_IMPL.key()}
+ OssOptions.OSS_IMPL.key(), OssOptions.OSS_PYTHON_IMPL.key(),
+ OssOptions.OSS_JINDO_PREAD_MAX_SIZE.key()}
+_DEFAULT_PREAD_MAX_SIZE =
OssOptions.OSS_JINDO_PREAD_MAX_SIZE.default_value().get_bytes()
+# Forward skips up to Jindo's read-ahead buffer stay on the stream, which
likely holds them.
+_STREAM_FORWARD_SKIP = 1024 * 1024
_CASE_SENSITIVE_JINDO_CONFIG_KEYS = {
OssOptions.OSS_ACCESS_KEY_ID.key().lower():
OssOptions.OSS_ACCESS_KEY_ID.key(),
OssOptions.OSS_ACCESS_KEY_SECRET.key().lower():
OssOptions.OSS_ACCESS_KEY_SECRET.key(),
@@ -167,13 +171,21 @@ def create_jindo_oss_filesystem(root_uri: str,
catalog_options: Options):
class JindoInputFile:
- def __init__(self, jindo_stream, file_size=None, *, owner):
+ def __init__(self, jindo_stream, file_size=None, *, owner,
+ pread_max_size=_DEFAULT_PREAD_MAX_SIZE):
self._stream = jindo_stream
self._file_size = file_size
# The stream does not own its native store: pin the jindo filesystem
(not the
# handler) so evicting the handler's FileIO from a cache cannot free
it.
self._owner = owner
self._closed = False
+ # PyArrow's ReadAt is seek + read: when enabled, small reads after a
long or
+ # backward jump use pread to skip the stream's read-ahead.
+ self._pread_max_size = pread_max_size
+ if pread_max_size > 0:
+ self._position = jindo_stream.tell() if hasattr(jindo_stream,
'tell') else 0
+ self._stream_position = self._position
+ self._last_read_end = None
_StreamStats.opened()
def __del__(self):
@@ -196,11 +208,48 @@ class JindoInputFile:
def read(self, nbytes: int = -1):
if self.closed:
raise ValueError("I/O operation on closed file")
+ if self._pread_max_size <= 0:
+ if nbytes is None or nbytes < 0:
+ if self._file_size is None:
+ return self._stream.read()
+ nbytes = max(0, self._file_size - self.tell())
+ return self._stream.read(nbytes)
if nbytes is None or nbytes < 0:
if self._file_size is None:
- return self._stream.read()
- nbytes = max(0, self._file_size - self.tell())
- return self._stream.read(nbytes)
+ self._sync_stream()
+ return self._advance(self._stream.read())
+ nbytes = max(0, self._file_size - self._position)
+ elif self._file_size is not None:
+ nbytes = max(0, min(nbytes, self._file_size - self._position))
+ if nbytes == 0:
+ return b""
+ if self._use_pread(nbytes):
+ data = self._stream.pread(nbytes, self._position)
+ self._position += len(data)
+ self._last_read_end = self._position
+ return data
+ self._sync_stream()
+ return self._advance(self._stream.read(nbytes))
+
+ def _use_pread(self, nbytes):
+ if (self._position == self._stream_position
+ or nbytes > self._pread_max_size
+ or not hasattr(self._stream, 'pread')):
+ return False
+ if self._last_read_end is None:
+ return True
+ return not 0 <= self._position - self._last_read_end <=
_STREAM_FORWARD_SKIP
+
+ def _sync_stream(self):
+ if self._stream_position != self._position:
+ self._stream.seek(self._position, os.SEEK_SET)
+ self._stream_position = self._position
+
+ def _advance(self, data):
+ self._position += len(data)
+ self._stream_position = self._position
+ self._last_read_end = self._position
+ return data
def seek(self, position: int, whence: int = 0):
if self.closed:
@@ -208,13 +257,29 @@ class JindoInputFile:
if whence == os.SEEK_END and self._file_size is not None:
position = self._file_size + position
whence = os.SEEK_SET
- self._stream.seek(position, whence)
- return self._stream.tell()
+ if self._pread_max_size <= 0:
+ self._stream.seek(position, whence)
+ return self._stream.tell()
+ if whence == os.SEEK_CUR:
+ position = self._position + position
+ elif whence == os.SEEK_END:
+ # Size unknown: let the native stream resolve SEEK_END.
+ self._stream.seek(position, whence)
+ self._stream_position = self._stream.tell()
+ position = self._stream_position
+ elif whence != os.SEEK_SET:
+ raise ValueError("Invalid whence: {}".format(whence))
+ if position < 0:
+ raise ValueError("Negative seek position {}".format(position))
+ self._position = position
+ return self._position
def tell(self) -> int:
if self.closed:
raise ValueError("I/O operation on closed file")
- return self._stream.tell()
+ if self._pread_max_size <= 0:
+ return self._stream.tell()
+ return self._position
def read_at(self, nbytes: int, offset: int):
if self.closed:
@@ -303,6 +368,8 @@ class JindoFileSystemHandler(FileSystemHandler):
self.properties = catalog_options
self._known_file_sizes = OrderedDict()
self._known_file_sizes_lock = threading.Lock()
+ self._pread_max_size = catalog_options.get(
+ OssOptions.OSS_JINDO_PREAD_MAX_SIZE).get_bytes()
config = build_jindo_config(catalog_options)
self._jindo_fs = jfs.connect(self.root_path, "root", config)
@@ -445,7 +512,7 @@ class JindoFileSystemHandler(FileSystemHandler):
jindo_stream = self._jindo_fs.open(normalized, "rb")
return PythonFile(
JindoInputFile(jindo_stream, self._known_file_size(normalized),
- owner=self._jindo_fs),
+ owner=self._jindo_fs,
pread_max_size=self._pread_max_size),
mode="r")
def open_input_file(self, path: str):
@@ -453,7 +520,7 @@ class JindoFileSystemHandler(FileSystemHandler):
jindo_stream = self._jindo_fs.open(normalized, "rb")
return PythonFile(
JindoInputFile(jindo_stream, self._known_file_size(normalized),
- owner=self._jindo_fs),
+ owner=self._jindo_fs,
pread_max_size=self._pread_max_size),
mode="r")
def open_output_stream(self, path: str, metadata):
diff --git a/paimon-python/pypaimon/tests/jindo_file_system_test.py
b/paimon-python/pypaimon/tests/jindo_file_system_test.py
index 96743573d0..e2df910cce 100644
--- a/paimon-python/pypaimon/tests/jindo_file_system_test.py
+++ b/paimon-python/pypaimon/tests/jindo_file_system_test.py
@@ -31,6 +31,7 @@ from unittest import mock
import pyarrow.fs as pafs
+from pyarrow import PythonFile
from pyarrow.fs import PyFileSystem
from pypaimon.catalog.rest.rest_token import RESTToken
from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO
@@ -80,6 +81,19 @@ class _RecordingInputStream:
self.closed = True
+_PREAD_MAX_SIZE = 4 * 1024 * 1024
+
+
+class _RecordingPreadStream(_RecordingInputStream):
+ def __init__(self, position=0):
+ super().__init__(position)
+ self.preads = []
+
+ def pread(self, size, offset):
+ self.preads.append((size, offset))
+ return b"x" * size
+
+
class JindoInputFileTest(unittest.TestCase):
def test_known_size_avoids_backend_seek_from_end(self):
@@ -96,6 +110,129 @@ class JindoInputFileTest(unittest.TestCase):
self.assertEqual(b"x" * 60, input_file.read())
self.assertEqual([60], stream.reads)
+ def test_read_after_seek_fetches_exact_range_with_pread(self):
+ stream = _RecordingPreadStream()
+ input_file = JindoInputFile(stream, file_size=10000, owner=object(),
pread_max_size=_PREAD_MAX_SIZE)
+ self.addCleanup(input_file.close)
+
+ input_file.seek(1000)
+ self.assertEqual(b"x" * 12, input_file.read(12))
+
+ self.assertEqual([(12, 1000)], stream.preads)
+ self.assertEqual([], stream.reads)
+ self.assertEqual([], stream.seeks)
+ self.assertEqual(1012, input_file.tell())
+
+ def test_read_continuing_after_pread_switches_to_stream(self):
+ stream = _RecordingPreadStream()
+ input_file = JindoInputFile(stream, file_size=10000, owner=object(),
pread_max_size=_PREAD_MAX_SIZE)
+ self.addCleanup(input_file.close)
+
+ input_file.seek(1000)
+ input_file.read(12)
+ input_file.read(20)
+ input_file.read(30)
+
+ self.assertEqual([(12, 1000)], stream.preads)
+ self.assertEqual([(1012, os.SEEK_SET)], stream.seeks)
+ self.assertEqual([20, 30], stream.reads)
+ self.assertEqual(1062, input_file.tell())
+
+ def test_sequential_reads_from_start_keep_stream_read(self):
+ stream = _RecordingPreadStream()
+ input_file = JindoInputFile(stream, file_size=10000, owner=object(),
pread_max_size=_PREAD_MAX_SIZE)
+ self.addCleanup(input_file.close)
+
+ input_file.read(10)
+ input_file.seek(10)
+ input_file.read(10)
+
+ self.assertEqual([], stream.preads)
+ self.assertEqual([], stream.seeks)
+ self.assertEqual([10, 10], stream.reads)
+
+ def test_short_forward_skip_keeps_stream(self):
+ stream = _RecordingPreadStream()
+ input_file = JindoInputFile(stream, file_size=10 * 1024 * 1024,
owner=object(), pread_max_size=_PREAD_MAX_SIZE)
+ self.addCleanup(input_file.close)
+
+ input_file.read(10)
+ input_file.seek(10 + 8192)
+ input_file.read(4)
+
+ self.assertEqual([], stream.preads)
+ self.assertEqual([(8202, os.SEEK_SET)], stream.seeks)
+ self.assertEqual([10, 4], stream.reads)
+
+ def test_long_or_backward_jump_uses_pread(self):
+ stream = _RecordingPreadStream()
+ input_file = JindoInputFile(stream, file_size=10 * 1024 * 1024,
owner=object(), pread_max_size=_PREAD_MAX_SIZE)
+ self.addCleanup(input_file.close)
+
+ input_file.read(10)
+ input_file.seek(2 * 1024 * 1024)
+ input_file.read(4)
+ input_file.seek(100)
+ input_file.read(4)
+
+ self.assertEqual([(4, 2 * 1024 * 1024), (4, 100)], stream.preads)
+ self.assertEqual([10], stream.reads)
+
+ def test_read_is_clamped_to_known_size(self):
+ stream = _RecordingPreadStream()
+ input_file = JindoInputFile(stream, file_size=100, owner=object(),
pread_max_size=_PREAD_MAX_SIZE)
+ self.addCleanup(input_file.close)
+
+ input_file.seek(95)
+ self.assertEqual(b"x" * 5, input_file.read(10))
+ self.assertEqual(b"", input_file.read(10))
+ self.assertEqual([(5, 95)], stream.preads)
+
+ def test_seek_from_current_position(self):
+ stream = _RecordingPreadStream()
+ input_file = JindoInputFile(stream, file_size=100, owner=object(),
pread_max_size=_PREAD_MAX_SIZE)
+ self.addCleanup(input_file.close)
+
+ input_file.read(10)
+ self.assertEqual(30, input_file.seek(20, os.SEEK_CUR))
+ self.assertEqual(30, input_file.tell())
+
+ def test_read_above_pread_max_size_keeps_stream(self):
+ stream = _RecordingPreadStream()
+ input_file = JindoInputFile(
+ stream, file_size=10000, owner=object(), pread_max_size=1000)
+ self.addCleanup(input_file.close)
+
+ input_file.seek(5000)
+ input_file.read(2000)
+
+ self.assertEqual([], stream.preads)
+ self.assertEqual([(5000, os.SEEK_SET)], stream.seeks)
+ self.assertEqual([2000], stream.reads)
+
+ def test_default_keeps_original_stream_behavior(self):
+ stream = _RecordingPreadStream()
+ input_file = JindoInputFile(stream, file_size=10000, owner=object())
+ self.addCleanup(input_file.close)
+
+ input_file.seek(5000)
+ self.assertEqual([(5000, os.SEEK_SET)], stream.seeks)
+ input_file.read(12)
+
+ self.assertEqual([], stream.preads)
+ self.assertEqual([12], stream.reads)
+ self.assertEqual(5012, input_file.tell())
+
+ def test_pyarrow_read_at_reaches_pread(self):
+ # PythonFile.read_at calls seek + read on the wrapped file.
+ stream = _RecordingPreadStream()
+ python_file = PythonFile(
+ JindoInputFile(stream, file_size=10000, owner=object(),
pread_max_size=_PREAD_MAX_SIZE), mode="r")
+
+ self.assertEqual(b"x" * 10, python_file.read_at(10, 500))
+ self.assertEqual([(10, 500)], stream.preads)
+ self.assertEqual([], stream.reads)
+
class JindoStreamStatsTest(unittest.TestCase):
@@ -223,6 +360,14 @@ class JindoConfigTest(unittest.TestCase):
self.assertNotIn(key, config.values)
self.assertEqual(config.values['fs.oss.endpoint'],
'oss-cn-test.example.com')
+ def test_pread_max_size_is_not_forwarded_to_jindo(self):
+ options = Options({'fs.oss.jindo.pread-max-size': '1mb',
+ 'fs.oss.endpoint': 'oss-cn-test.example.com'})
+ with mock.patch.object(jindo_module, 'JINDO_AVAILABLE', True), \
+ mock.patch.object(jindo_module, 'jutil',
types.SimpleNamespace(Config=_RecordingConfig)):
+ config = jindo_module.build_jindo_config(options)
+ self.assertNotIn('fs.oss.jindo.pread-max-size', config.values)
+
def test_forwards_native_options_to_connect(self):
created_config = _RecordingConfig()
config_factory = mock.Mock(return_value=created_config)
@@ -476,6 +621,27 @@ class JindoKeepAliveTest(unittest.TestCase):
handler = JindoFileSystemHandler("oss://bucket/", options)
return handler, weakref.ref(fake_fs), weakref.ref(handler)
+ def test_handler_passes_pread_max_size_to_input_files(self):
+ options = Options({
+ "fs.oss.impl": "jindo",
+ "fs.oss.endpoint": "https://oss-cn-hangzhou.aliyuncs.com";,
+ "fs.oss.jindo.pread-max-size": "1kb",
+ })
+ fake_jfs = types.SimpleNamespace(
+ connect=lambda *a, **kw: _EvictionRecordingFileSystem([]))
+ with mock.patch.object(jindo_module, "JINDO_AVAILABLE", True), \
+ mock.patch.object(jindo_module, "jfs", fake_jfs), \
+ mock.patch.object(jindo_module, "jutil",
+
types.SimpleNamespace(Config=_RecordingConfig)):
+ handler = JindoFileSystemHandler("oss://bucket/", options)
+ with mock.patch.object(jindo_module, "JindoInputFile",
+ wraps=JindoInputFile) as input_file_class:
+ handler.open_input_file('object').close()
+ handler.open_input_stream('object').close()
+ for call in input_file_class.call_args_list:
+ self.assertEqual(1024, call.kwargs['pread_max_size'])
+ self.assertEqual(2, input_file_class.call_count)
+
def test_evicted_filesystem_outlives_open_stream(self):
events = []
handler, fs_ref, handler_ref = self._handler_with_recording_fs(events)
@@ -610,6 +776,25 @@ class JindoTokenRefreshTest(unittest.TestCase):
stream.close()
self.assertEqual(["stream-closed"], events)
+ def test_data_token_overrides_pread_max_size(self):
+ table_path = "oss://bucket/db.db/t"
+ token = self._token("sts-1", int(time.time() * 1000) + 7_200_000)
+ token.token["fs.oss.jindo.pread-max-size"] = "2mb"
+ with mock.patch.object(jindo_module, "JINDO_AVAILABLE", True), \
+ mock.patch.object(pyarrow_file_io_module, "JINDO_AVAILABLE",
True), \
+ mock.patch.object(jindo_module, "jfs", types.SimpleNamespace(
+ connect=lambda *a, **kw:
_EvictionRecordingFileSystem([]))), \
+ mock.patch.object(jindo_module, "jutil",
+
types.SimpleNamespace(Config=_RecordingConfig)), \
+ mock.patch.object(RESTTokenFileIO, "try_to_refresh_token"):
+ reader = RESTTokenFileIO(
+ Identifier.create("db", "t"), table_path,
+ Options({OssOptions.OSS_IMPL.key(): "jindo",
+ "fs.oss.jindo.pread-max-size": "1mb"}))
+ reader.token = token
+ reader.file_io()
+ self.assertEqual(2 * 1024 * 1024,
reader.filesystem.handler._pread_max_size)
+
class JindoFileSystemTest(unittest.TestCase):
"""Test cases for JindoFileSystem."""
