[python] Back-reference keeping the handler alive in stream for pyjindo impl

paimon-python/pypaimon/filesystem/jindo_file_system_handler.py

@@ -131,11 +131,31 @@ def create_jindo_oss_filesystem(root_uri: str, catalog_options: Options):
 
 
 class JindoInputFile:
-    def __init__(self, jindo_stream, file_size=None):
+    def __init__(self, jindo_stream, file_size=None, *, owner):
         self._stream = jindo_stream
         self._file_size = file_size
+        # Keep-alive back-reference to the owning jindo filesystem: the
+        # stream is a non-owning reference into memory owned by it. Without
+        # it, dropping the FileIO that holds the handler (e.g.
+        # RESTTokenFileIO's TTLCache maxsize=1 eviction on token refresh)
+        # frees the native connection while other threads still read from
+        # this stream. Keyword-only and required so a new open_* path cannot
+        # silently skip it; it pins only the filesystem, not the handler, so
+        # the evicted FileIO's properties do not stay retained.
+        self._owner = owner
         self._closed = False
 
+    def __del__(self):
+        # CPython releases instance attributes in dict layout order, which is
+        # an implementation detail, not a contract. Close the native stream
+        # while ``_owner`` still pins the store so the stream's own release
+        # cannot race the store's teardown (pyjindo's JindoOssFile relies on
+        # the same guarantee via io.IOBase's finalizer).
+        try:
+            self.close()
+        except Exception:
+            pass
+
     @property
     def closed(self):
         if hasattr(self._stream, 'closed'):
@@ -174,6 +194,9 @@ def close(self):
         if not self._closed:
             self._stream.close()
             self._closed = True
+            # The native stream is closed: a file object that stays
+            # referenced no longer needs to pin the native store.
+            self._owner = None
 
     def __enter__(self):
         return self
@@ -184,10 +207,21 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
 
 class JindoOutputFile:
-    def __init__(self, jindo_stream):
+    def __init__(self, jindo_stream, *, owner):
         self._stream = jindo_stream
+        # See JindoInputFile._owner: keep the native store alive while this
+        # stream is open.
+        self._owner = owner
         self._closed = False
 
+    def __del__(self):
+        # See JindoInputFile.__del__: close while ``_owner`` still pins the
+        # store. For output streams this also flushes pending writes.
+        try:
+            self.close()
+        except Exception:
+            pass
+
     @property
     def closed(self):
         if hasattr(self._stream, 'closed'):
@@ -213,6 +247,9 @@ def close(self):
         if not self._closed:
             self._stream.close()
             self._closed = True
+            # The native stream is closed: a file object that stays
+            # referenced no longer needs to pin the native store.
+            self._owner = None
 
     def __enter__(self):
         return self
@@ -373,20 +410,22 @@ def open_input_stream(self, path: str):
         normalized = self._normalize_path(path)
         jindo_stream = self._jindo_fs.open(normalized, "rb")
         return PythonFile(
-            JindoInputFile(jindo_stream, self._known_file_size(normalized)),
+            JindoInputFile(jindo_stream, self._known_file_size(normalized),
+                           owner=self._jindo_fs),
             mode="r")
 
     def open_input_file(self, path: str):
         normalized = self._normalize_path(path)
         jindo_stream = self._jindo_fs.open(normalized, "rb")
         return PythonFile(
-            JindoInputFile(jindo_stream, self._known_file_size(normalized)),
+            JindoInputFile(jindo_stream, self._known_file_size(normalized),
+                           owner=self._jindo_fs),
             mode="r")
 
     def open_output_stream(self, path: str, metadata):
         normalized = self._normalize_path(path)
         jindo_stream = self._jindo_fs.open(normalized, "wb")
-        return PythonFile(JindoOutputFile(jindo_stream), mode="w")
+        return PythonFile(JindoOutputFile(jindo_stream, owner=self._jindo_fs), mode="w")
 
     def open_append_stream(self, path: str, metadata):
         raise IOError("append mode is not supported")

paimon-python/pypaimon/tests/jindo_file_system_test.py

@@ -15,13 +15,15 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import gc
 import io
 import os
 import pickle
 import threading
 import types
 import unittest
 import uuid
+import weakref
 from unittest import mock
 
 import pyarrow.fs as pafs
@@ -34,6 +36,7 @@
 from pypaimon.filesystem.jindo_file_system_handler import (
     JindoFileSystemHandler,
     JindoInputFile,
+    JindoOutputFile,
     JINDO_AVAILABLE,
 )
 
@@ -74,14 +77,14 @@ class JindoInputFileTest(unittest.TestCase):
 
     def test_known_size_avoids_backend_seek_from_end(self):
         stream = _RecordingInputStream()
-        input_file = JindoInputFile(stream, file_size=100)
+        input_file = JindoInputFile(stream, file_size=100, owner=object())
 
         self.assertEqual(95, input_file.seek(-5, os.SEEK_END))
         self.assertEqual([(95, os.SEEK_SET)], stream.seeks)
 
     def test_known_size_bounds_unlimited_read(self):
         stream = _RecordingInputStream(position=40)
-        input_file = JindoInputFile(stream, file_size=100)
+        input_file = JindoInputFile(stream, file_size=100, owner=object())
 
         self.assertEqual(b"x" * 60, input_file.read())
         self.assertEqual([60], stream.reads)
@@ -101,7 +104,7 @@ def seek(self, offset, whence=io.SEEK_SET):
             def tell(self):
                 return self._stream.tell()
 
-        stream = JindoInputFile(JindoStream())
+        stream = JindoInputFile(JindoStream(), owner=object())
         self.assertEqual(5, stream.seek(0, io.SEEK_END))
 
     def test_pickle_reconnects_without_native_client_lock_or_cache(self):
@@ -326,6 +329,165 @@ def test_forwards_native_options_to_jindo_oss_filesystem(self):
         self.assertEqual(created_config.values["logger.verbose"], "3")
 
 
+class _EvictionRecordingFileSystem:
+    """Fake ``jfs.connect`` result: owns the native store, records teardown."""
+
+    def __init__(self, events):
+        self.events = events
+
+    def open(self, path, mode):
+        return _EvictionRecordingNativeStream(self, self.events)
+
+    def __del__(self):
+        self.events.append('fs-destroyed')
+
+
+class _EvictionRecordingNativeStream:
+    """Fake pyjindo stream: non-owning reference to the native store."""
+
+    def __init__(self, fake_fs, events):
+        # Weak on purpose: the native stream must not keep the store alive.
+        self._fs = weakref.ref(fake_fs)
+        self._events = events
+        self.closed = False
+
+    def read(self, size=None):
+        if self._fs() is None:
+            raise RuntimeError('use-after-free: read after store teardown')
+        return b'x' * (size or 1)
+
+    def pread(self, size, offset):
+        return self.read(size)
+
+    def write(self, data):
+        if self._fs() is None:
+            raise RuntimeError('use-after-free: write after store teardown')
+        return len(data)
+
+    def seek(self, position, whence=0):
+        return position
+
+    def tell(self):
+        return 0
+
+    def close(self):
+        if self.closed:
+            return
+        if self._fs() is None:
+            # Emulates the native use-after-free the keep-alive prevents.
+            self._events.append('stream-closed-after-store-destroyed')
+            return
+        self.closed = True
+        self._events.append('stream-closed')
+
+    def __del__(self):
+        # pyjindo closes its native streams when they are released.
+        if not self.closed:
+            self.close()
+
+
+class JindoKeepAliveTest(unittest.TestCase):
+    """Streams must keep the native store alive across FileIO eviction
+    (RESTTokenFileIO token refresh) and must close before the store is
+    released, without relying on attribute dealloc order."""
+
+    @staticmethod
+    def _handler_with_recording_fs(events):
+        options = Options({
+            "fs.oss.impl": "jindo",
+            "fs.oss.endpoint": "https://oss-cn-hangzhou.aliyuncs.com",
+            "fs.oss.accessKeyId": "test-ak",
+            "fs.oss.accessKeySecret": "test-sk",
+        })
+        fake_fs = _EvictionRecordingFileSystem(events)
+        fake_jfs = types.SimpleNamespace(connect=lambda *a, **kw: fake_fs)
+        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)
+        return handler, weakref.ref(fake_fs), weakref.ref(handler)
+
+    def test_evicted_filesystem_outlives_open_stream(self):
+        events = []
+        handler, fs_ref, handler_ref = self._handler_with_recording_fs(events)
+        input_file = handler.open_input_file('object')
+
+        # TTLCache eviction on token refresh drops the handler chain: the
+        # handler (and its credentials) must die, the native store must not.
+        del handler
+        gc.collect()
+        self.assertIsNone(
+            handler_ref(), 'eviction must not pin the handler')
+        self.assertIsNotNone(
+            fs_ref(), 'eviction freed the native store while the stream was open')
+
+        # The stream keeps reading after eviction.
+        self.assertEqual(b'x', input_file.read(1))
+
+        # Closing releases the keep-alive: the still-referenced closed file
+        # no longer pins the store, and the stream closes before teardown.
+        input_file.close()
+        self.assertEqual(['stream-closed', 'fs-destroyed'], events)
+        self.assertIsNone(fs_ref())
+        del input_file
+        gc.collect()
+        # No double close on finalization.
+        self.assertEqual(['stream-closed', 'fs-destroyed'], events)
+
+    def test_evicted_filesystem_outlives_open_output_stream(self):
+        events = []
+        handler, fs_ref, handler_ref = self._handler_with_recording_fs(events)
+        output_file = handler.open_output_stream('object', None)
+
+        del handler
+        gc.collect()
+        self.assertIsNone(handler_ref())
+        self.assertIsNotNone(fs_ref())
+
+        self.assertEqual(4, output_file.write(b'data'))
+        output_file.close()
+        self.assertEqual(['stream-closed', 'fs-destroyed'], events)
+        self.assertIsNone(fs_ref())
+
+    def test_finalizer_closes_stream_before_store_teardown(self):
+        # Attribute release order is a CPython implementation detail: if a
+        # future refactor assigns _owner before _stream in __init__, the
+        # owner would be released first on dealloc. The finalizer must still
+        # close the stream while the owner pins the store.
+        events = []
+        fs = _EvictionRecordingFileSystem(events)
+        fs_ref = weakref.ref(fs)
+        stream = _EvictionRecordingNativeStream(fs, events)
+
+        class _ReversedInitInputFile(JindoInputFile):
+            def __init__(self, jindo_stream, owner):
+                self._owner = owner
+                self._stream = jindo_stream
+                self._file_size = None
+                self._closed = False
+
+        input_file = _ReversedInitInputFile(stream, owner=fs)
+        del fs, stream
+        gc.collect()
+        self.assertIsNotNone(fs_ref())
+
+        # Drop without closing: the finalizer must close the stream first...
+        del input_file
+        gc.collect()
+        self.assertIsNone(fs_ref())
+        # ...and the store must be destroyed only afterwards.
+        self.assertEqual(
+            ['stream-closed', 'fs-destroyed'], events,
+            'native stream must be closed before the store is destroyed')
+
+    def test_owner_argument_is_required(self):
+        with self.assertRaises(TypeError):
+            JindoInputFile(_RecordingInputStream())
+        with self.assertRaises(TypeError):
+            JindoOutputFile(_RecordingInputStream())
+
+
 class JindoFileSystemTest(unittest.TestCase):
     """Test cases for JindoFileSystem."""