[Stephen0421]

Purpose

CachingInputStream.read_at is documented as thread-safe, but the lazy open behind it was not:

def _get_remote_stream(self):
    if self._stream is None:
        self._stream = self._file_io.new_input_stream(self._file_path)
    return self._stream

On a cold cache, several threads reading the same stream can all observe self._stream is None and each open a remote stream. Only the last assignment is retained, so close() closes one and leaks the rest.

This is reachable as soon as local cache is enabled, with no further options. local-cache.enabled defaults to false, so a stock table never enters this path. When the cache is on, global-index is already in the default whitelist meta,global-index, and _read_parallelism defaults to 4 for remote schemes. PaimonVindexInput.pread_many then fans
ranges onto one shared CachingInputStream.

Read results were never wrong — the actual I/O is still serialized by _io_lock or executed through pread. What leaked was object store connections.

Java already guards the same lazy open in CachingSeekableInputStream.getRemoteStream, for the same reason:

// reached concurrently: readVectored fans preadFully out over an IO thread pool, so an
// unguarded lazy init would open one stream per thread and leak all but the last

The open uses its own _open_lock. _get_remote_stream runs before _io_lock, and _io_lock only covers seek + read after a stream already exists. Folding the open into _io_lock would also be race-free, but that lock means the cursor is in use, and there is no stream yet while it is being opened. CachingSeekableInputStream makes the same split: a
private init lock, distinct from synchronized (stream).

With 16 threads reading disjoint cold offsets, the number of opened streams drops from 16 to 1, and close() leaves none behind.

Not addressed here

Two related gaps are left for separate changes, to keep this fix focused:

• close() racing an in-flight open. Java additionally tracks a closed flag and has the opening thread re-check it after publishing, closing the stream it just opened if close() won. PyPaimon has no such flag. The window is narrower here because PaimonVindexInput.close shuts its executor down with wait=True, draining in-flight reads before
  VindexReaderResources.close closes the underlying stream. Adding the flag changes the post-close read behaviour, which deserves its own review.
• _get_file_size is also unguarded. Concurrent callers can duplicate the stat, but it neither leaks connections nor returns wrong data.

Tests

CachingInputStreamTest.test_concurrent_read_at_opens_one_remote_stream runs 16 threads over disjoint cold offsets and asserts exactly one remote stream is opened, that every read returns the correct bytes, and that close() leaves no stream open.

The test fails on the unfixed code (16 streams opened) and passes with the fix.