[python] Make BLOB I/O merge parameters configurable

paimon-python/README.md

@@ -133,6 +133,25 @@ and precomputed primary-key global-index results still use the Python planner.
 Continuous streaming and write planning also retain their Python entrypoints.
 Native planning remains optional and is disabled by default.
 
+# Coalesced BLOB reads
+
+FileIO merges nearby BLOB ranges before reading. Set
+`file-io.read-coalesce.max-gap` and `file-io.read-coalesce.max-block` in the
+catalog or connection options to tune the 1 MiB and 8 MiB defaults:
+
+```python
+import pypaimon.multimodal as pmm
+
+connection = pmm.connect(options={
+    "warehouse": "/tmp/warehouse",
+    "file-io.read-coalesce.max-gap": "64 kb",
+    "file-io.read-coalesce.max-block": "16 mb",
+})
+```
+
+`max-block` constrains coalescing, but does not split an individual BLOB range.
+A single read can therefore exceed this value.
+
 # Load LeRobot Dataset v3
 
 Install the optional dependency, then import a local directory, FileIO URI, or

paimon-python/pypaimon/common/file_io.py

@@ -50,10 +50,6 @@ def pread(stream, length: int, offset: int) -> bytes:
     return os.pread(stream.fileno(), length, offset)
 
 
-# Coalescing bounds: merge same-file ranges whose gap is within GAP, capping a
-# merged read at SPAN so threads stay busy and memory stays bounded.
-_COALESCE_GAP = 1 << 20
-_COALESCE_SPAN = 8 << 20
 _COALESCE_VIEW_MAX_RETAINED_AMPLIFICATION = 2.0
 # Bound per-object opens; 16 cuts them by 75% for default 64-range batches.
 _MAX_RANGE_LANES_PER_PATH = 16
@@ -187,8 +183,7 @@ def read_file_range(self, path, offset, length):
         finally:
             stream.close()
 
-    def read_ranges_coalesced(self, ranges, parallelism,
-                              max_gap=_COALESCE_GAP, max_span=_COALESCE_SPAN):
+    def read_ranges_coalesced(self, ranges, parallelism):
         """Read ``ranges`` (each ``None`` or ``(path, offset, length)``), returning
         bytes in the same order. Same-file nearby ranges are merged into one read
         to cut round trips, then sliced. Each worker lane reuses one exclusive
@@ -198,12 +193,12 @@ def read_ranges_coalesced(self, ranges, parallelism,
         A failed read propagates and aborts the whole batch (unlike a per-row
         ``file.open()`` loop that fails one row at a time).
         """
+        max_gap, max_span = self._resolve_coalesce_limits()
         return self._read_ranges_coalesced(
             ranges, parallelism, max_gap, max_span,
             max_retained_amplification=0, return_views=False)
 
-    def read_ranges_coalesced_views(self, ranges, parallelism,
-                                    max_gap=_COALESCE_GAP, max_span=_COALESCE_SPAN,
+    def read_ranges_coalesced_views(self, ranges, parallelism, *,
                                     max_retained_amplification=(
                                         _COALESCE_VIEW_MAX_RETAINED_AMPLIFICATION)):
         """Read coalesced ranges as zero-copy ``memoryview`` slices.
@@ -216,10 +211,21 @@ def read_ranges_coalesced_views(self, ranges, parallelism,
         excessive gap bytes; set ``max_retained_amplification`` to a non-positive
         value to always share the merged buffer.
         """
+        max_gap, max_span = self._resolve_coalesce_limits()
         return self._read_ranges_coalesced(
             ranges, parallelism, max_gap, max_span, max_retained_amplification,
             return_views=True)
 
+    def _resolve_coalesce_limits(self):
+        from pypaimon.common.options.config import FileIOOptions
+        properties = getattr(self, "properties", None)
+        if not isinstance(properties, Options):
+            properties = Options({})
+        return (
+            properties.get(FileIOOptions.READ_COALESCE_MAX_GAP).get_bytes(),
+            properties.get(FileIOOptions.READ_COALESCE_MAX_BLOCK).get_bytes(),
+        )
+
     def _read_ranges_coalesced(self, ranges, parallelism, max_gap, max_span,
                                max_retained_amplification, return_views):
         from concurrent.futures import ThreadPoolExecutor

paimon-python/pypaimon/common/options/config.py

@@ -134,6 +134,26 @@ class CatalogOptions:
     BLOB_FILE_IO_DEFAULT_CACHE_SIZE = 2 ** 31 - 1
 
 
+class FileIOOptions:
+    READ_COALESCE_MAX_GAP = (
+        ConfigOptions.key("file-io.read-coalesce.max-gap")
+        .memory_type()
+        .default_value(MemorySize.of_mebi_bytes(1))
+        .with_description(
+            "Maximum gap between same-file ranges merged into one read."
+        )
+    )
+    READ_COALESCE_MAX_BLOCK = (
+        ConfigOptions.key("file-io.read-coalesce.max-block")
+        .memory_type()
+        .default_value(MemorySize.of_mebi_bytes(8))
+        .with_description(
+            "Maximum span for coalescing same-file ranges, except when an "
+            "individual range is larger. Individual ranges are not split."
+        )
+    )
+
+
 class HdfsOptions:
     HDFS_CLIENT_IMPL = (
         ConfigOptions.key("hdfs.client.impl")