[python] Prefer Jindo for native scan planning

paimon-python/pypaimon/read/native_plan.py

@@ -24,7 +24,7 @@
 
 from typing import List, Optional, Tuple
 
-from pypaimon.common.options.config import CatalogOptions
+from pypaimon.common.options.config import CatalogOptions, OssOptions
 from pypaimon.common.options.core_options import CoreOptions
 from pypaimon.common.options.options_utils import OptionsUtils
 from pypaimon.common.predicate import Predicate
@@ -93,6 +93,14 @@ def _catalog_options(table) -> dict:
     if metastore is None:
         raise ValueError("native_plan requires an exact built-in catalog loader")
     normalized[CatalogOptions.METASTORE.key()] = metastore
+    if str(getattr(table, 'table_path', '')).startswith('oss://'):
+        from pypaimon.filesystem.jindo_file_system_handler import (
+            JINDO_AVAILABLE,
+        )
+        impl = normalized.get(OssOptions.OSS_IMPL.key())
+        if JINDO_AVAILABLE and (impl is None or impl.lower() == 'jindo'):
+            # This catalog is only used for Rust scan planning.
+            normalized[OssOptions.OSS_IMPL.key()] = 'jindo'
     return normalized
 
 

paimon-python/pypaimon/tests/native_plan_test.py

@@ -425,6 +425,43 @@ def test_catalog_options_use_actual_rest_loader_type(self):
             'metastore': 'rest',
         })
 
+    @patch(
+        'pypaimon.filesystem.jindo_file_system_handler.JINDO_AVAILABLE', True)
+    def test_native_plan_prefers_installed_jindo_for_oss(self):
+        table = Mock(table_path='oss://bucket/table')
+        table.catalog_environment.catalog_loader = FileSystemCatalogLoader(
+            CatalogContext.create_from_options(Options({})))
+
+        self.assertEqual(_catalog_options(table), {
+            'metastore': 'filesystem',
+            'fs.oss.impl': 'jindo',
+        })
+
+    @patch(
+        'pypaimon.filesystem.jindo_file_system_handler.JINDO_AVAILABLE', False)
+    def test_native_plan_uses_opendal_without_jindo(self):
+        table = Mock(table_path='oss://bucket/table')
+        table.catalog_environment.catalog_loader = FileSystemCatalogLoader(
+            CatalogContext.create_from_options(Options({})))
+
+        self.assertEqual(_catalog_options(table), {
+            'metastore': 'filesystem',
+        })
+
+    @patch(
+        'pypaimon.filesystem.jindo_file_system_handler.JINDO_AVAILABLE', True)
+    def test_native_plan_respects_explicit_legacy_oss(self):
+        table = Mock(table_path='oss://bucket/table')
+        table.catalog_environment.catalog_loader = FileSystemCatalogLoader(
+            CatalogContext.create_from_options(Options({
+                'fs.oss.impl': 'legacy',
+            })))
+
+        self.assertEqual(_catalog_options(table), {
+            'fs.oss.impl': 'legacy',
+            'metastore': 'filesystem',
+        })
+
     def test_catalog_options_reject_loader_subclass(self):
         class RoutedFileSystemLoader(FileSystemCatalogLoader):
             pass