[fix](fe) Address Paimon JDBC review feedback

### What problem does this PR solve?

Issue Number: None

Related PR: #61513

Problem Summary: Address review feedback for Paimon JDBC JNI scans by moving backend Paimon options to scan-level params with BE fallback, aligning driver_class validation exceptions, cleaning up JDBC driver test state, and removing the unrelated build change from the PR branch history.

### Release note

None

### Check List (For Author)

- Test: Unit Test / Static check

    - git diff --check on the staged files

    - ./run-fe-ut.sh --run org.apache.doris.datasource.property.metastore.PaimonJdbcMetaStorePropertiesTest,org.apache.doris.datasource.paimon.source.PaimonScanNodeTest,org.apache.doris.paimon.PaimonJdbcDriverUtilsTest (still running in the PR worktree at commit time)

- Behavior changed: Yes (Paimon backend options now propagate at scan level with BE fallback; validation exception type is aligned)

- Does this need documentation: No

Author: xylaaaaa

be/src/format/table/paimon_cpp_reader.cpp

@@ -269,8 +269,10 @@ std::vector<std::string> PaimonCppReader::_build_read_columns() const {
 
 std::map<std::string, std::string> PaimonCppReader::_build_options() const {
     std::map<std::string, std::string> options;
-    if (_range.__isset.table_format_params && _range.table_format_params.__isset.paimon_params &&
-        _range.table_format_params.paimon_params.__isset.paimon_options) {
+    if (_range_params && _range_params->__isset.paimon_options && !_range_params->paimon_options.empty()) {
+        options.insert(_range_params->paimon_options.begin(), _range_params->paimon_options.end());
+    } else if (_range.__isset.table_format_params && _range.table_format_params.__isset.paimon_params &&
+               _range.table_format_params.paimon_params.__isset.paimon_options) {
         options.insert(_range.table_format_params.paimon_params.paimon_options.begin(),
                        _range.table_format_params.paimon_params.paimon_options.end());
     }
@@ -310,7 +312,6 @@ std::map<std::string, std::string> PaimonCppReader::_build_options() const {
     copy_if_missing("fs.s3a.region", "AWS_REGION");
     copy_if_missing("fs.s3a.path.style.access", "use_path_style");
 
-    // FE currently does not pass paimon_options in scan ranges.
     // Backfill file.format/manifest.format from split file_format to avoid
     // paimon-cpp falling back to default manifest.format=avro.
     if (_range.__isset.table_format_params && _range.table_format_params.__isset.paimon_params &&

be/src/format/table/paimon_jni_reader.cpp

@@ -65,8 +65,15 @@ PaimonJniReader::PaimonJniReader(const std::vector<SlotDescriptor*>& file_slot_d
                       if (range_params->__isset.serialized_table) {
                           params["serialized_table"] = range_params->serialized_table;
                       }
-                      for (const auto& kv : paimon_params.paimon_options) {
-                          params[PAIMON_OPTION_PREFIX + kv.first] = kv.second;
+                      if (range_params->__isset.paimon_options &&
+                          !range_params->paimon_options.empty()) {
+                          for (const auto& kv : range_params->paimon_options) {
+                              params[PAIMON_OPTION_PREFIX + kv.first] = kv.second;
+                          }
+                      } else if (paimon_params.__isset.paimon_options) {
+                          for (const auto& kv : paimon_params.paimon_options) {
+                              params[PAIMON_OPTION_PREFIX + kv.first] = kv.second;
+                          }
                       }
                       if (range_params->__isset.properties && !range_params->properties.empty()) {
                           for (const auto& kv : range_params->properties) {

fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJdbcDriverUtilsTest.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.common.classloader.JniScannerClassLoader;
 
+import org.junit.After;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -30,6 +31,7 @@
 import java.sql.DriverManager;
 import java.sql.DriverPropertyInfo;
 import java.sql.SQLFeatureNotSupportedException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -39,17 +41,34 @@
 import java.util.logging.Logger;
 
 public class PaimonJdbcDriverUtilsTest {
+    private final List<Driver> registeredDrivers = new ArrayList<>();
+    private final List<Path> tempJars = new ArrayList<>();
+
+    @After
+    public void tearDown() throws Exception {
+        for (Driver driver : registeredDrivers) {
+            DriverManager.deregisterDriver(driver);
+        }
+        registeredDrivers.clear();
+        for (Path tempJar : tempJars) {
+            Files.deleteIfExists(tempJar);
+        }
+        tempJars.clear();
+    }
+
     @Test
     public void testRegisterDriverIfNeeded() throws Exception {
         Path driverJar = createDriverJar();
         Map<String, String> params = new HashMap<>();
         params.put(PaimonJdbcDriverUtils.PAIMON_JDBC_DRIVER_URL, driverJar.toUri().toURL().toString());
         params.put(PaimonJdbcDriverUtils.PAIMON_JDBC_DRIVER_CLASS, DummyJdbcDriver.class.getName());
 
-        JniScannerClassLoader scannerClassLoader = new JniScannerClassLoader("paimon-test", List.of(), null);
+        JniScannerClassLoader scannerClassLoader =
+                new JniScannerClassLoader("paimon-test", List.of(), ClassLoader.getPlatformClassLoader());
         PaimonJdbcDriverUtils.registerDriverIfNeeded(params, scannerClassLoader);
 
         Driver driver = DriverManager.getDriver("jdbc:dummy:test");
+        registeredDrivers.add(driver);
         Assert.assertTrue(driver.acceptsURL("jdbc:dummy:test"));
     }
 
@@ -65,6 +84,7 @@ public void testRegisterDriverIfNeededRequiresDriverClass() {
 
     private Path createDriverJar() throws IOException {
         Path jarPath = Files.createTempFile("paimon-jdbc-driver", ".jar");
+        tempJars.add(jarPath);
         String resourceName = DummyJdbcDriver.class.getName().replace('.', '/') + ".class";
         try (JarOutputStream jarOutputStream = new JarOutputStream(Files.newOutputStream(jarPath));
                 InputStream inputStream = DummyJdbcDriver.class.getClassLoader().getResourceAsStream(resourceName)) {

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java

@@ -205,6 +205,13 @@ public void createScanRangeLocations() throws UserException {
         // Set paimon_predicate at ScanNode level to avoid redundant serialization in each split
         String serializedPredicate = PaimonUtil.encodeObjectToString(predicates);
         params.setPaimonPredicate(serializedPredicate);
+        setScanLevelPaimonOptions();
+    }
+
+    private void setScanLevelPaimonOptions() {
+        if (!backendPaimonOptions.isEmpty()) {
+            params.setPaimonOptions(backendPaimonOptions);
+        }
     }
 
     private void putHistorySchemaInfo(Long schemaId) {
@@ -261,9 +268,6 @@ private void setPaimonParams(TFileRangeDesc rangeDesc, PaimonSplit paimonSplit)
             fileDesc.setSchemaId(paimonSplit.getSchemaId());
         }
         fileDesc.setFileFormat(fileFormat);
-        if (!backendPaimonOptions.isEmpty()) {
-            fileDesc.setPaimonOptions(backendPaimonOptions);
-        }
         // Hadoop conf is set at ScanNode level via params.properties in createScanRangeLocations(),
         // no need to set it for each split to avoid redundant configuration
         Optional<DeletionFile> optDeletionFile = paimonSplit.getDeletionFile();

fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/PaimonJdbcMetaStoreProperties.java

@@ -166,7 +166,7 @@ public Map<String, String> getBackendPaimonOptions() {
             return Collections.emptyMap();
         }
         if (StringUtils.isBlank(driverClass)) {
-            throw new IllegalStateException("jdbc.driver_class or paimon.jdbc.driver_class is required when "
+            throw new IllegalArgumentException("jdbc.driver_class or paimon.jdbc.driver_class is required when "
                     + "jdbc.driver_url or paimon.jdbc.driver_url is specified");
         }
         Map<String, String> backendPaimonOptions = new HashMap<>();

fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java

@@ -32,6 +32,8 @@
 import org.apache.doris.planner.PlanNodeId;
 import org.apache.doris.planner.ScanContext;
 import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.thrift.TFileRangeDesc;
+import org.apache.doris.thrift.TFileScanRangeParams;
 
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.io.DataFileMeta;
@@ -511,11 +513,67 @@ public void testGetBackendPaimonOptionsForJdbcCatalog() throws Exception {
         Assert.assertEquals(2, backendOptions.size());
     }
 
+    @Test
+    public void testApplyBackendPaimonOptionsAtScanNodeLevel() throws Exception {
+        PaimonScanNode node = new PaimonScanNode(new PlanNodeId(0), new TupleDescriptor(new TupleId(0)),
+                false, sv, ScanContext.EMPTY);
+        PaimonSource source = Mockito.mock(PaimonSource.class);
+        Mockito.when(source.getTableLocation()).thenReturn("file:///warehouse");
+        node.setSource(source);
+
+        Map<String, String> backendOptions = new HashMap<>();
+        backendOptions.put("jdbc.driver_url", "file:///tmp/postgresql-42.5.0.jar");
+        backendOptions.put("jdbc.driver_class", "org.postgresql.Driver");
+        setField(FileQueryScanNode.class, node, "params", new TFileScanRangeParams());
+        setField(PaimonScanNode.class, node, "backendPaimonOptions", backendOptions);
+        setField(PaimonScanNode.class, node, "storagePropertiesMap", Collections.emptyMap());
+
+        invokePrivateMethod(node, "setScanLevelPaimonOptions");
+
+        Assert.assertEquals(backendOptions, node.getFileScanRangeParams().getPaimonOptions());
+
+        TFileRangeDesc rangeDesc = new TFileRangeDesc();
+        invokePrivateMethod(node, "setPaimonParams",
+                new Class<?>[] {TFileRangeDesc.class, PaimonSplit.class},
+                rangeDesc, new PaimonSplit(createDataSplit("scan_level.parquet")));
+        Assert.assertFalse(rangeDesc.getTableFormatParams().getPaimonParams().isSetPaimonOptions());
+    }
+
     private void mockJniReader(PaimonScanNode spyNode) {
         Mockito.doReturn(false).when(spyNode).supportNativeReader(ArgumentMatchers.any(Optional.class));
     }
 
     private void mockNativeReader(PaimonScanNode spyNode) {
         Mockito.doReturn(true).when(spyNode).supportNativeReader(ArgumentMatchers.any(Optional.class));
     }
+
+    private void setField(Class<?> clazz, Object target, String fieldName, Object value) throws Exception {
+        java.lang.reflect.Field field = clazz.getDeclaredField(fieldName);
+        field.setAccessible(true);
+        field.set(target, value);
+    }
+
+    private Object invokePrivateMethod(Object target, String methodName, Class<?>[] parameterTypes, Object... args)
+            throws Exception {
+        Method method = target.getClass().getDeclaredMethod(methodName, parameterTypes);
+        method.setAccessible(true);
+        return method.invoke(target, args);
+    }
+
+    private Object invokePrivateMethod(Object target, String methodName) throws Exception {
+        return invokePrivateMethod(target, methodName, new Class<?>[0]);
+    }
+
+    private DataSplit createDataSplit(String fileName) {
+        DataFileMeta dataFileMeta = DataFileMeta.forAppend(fileName, 64L * 1024 * 1024, 1L, SimpleStats.EMPTY_STATS,
+                1L, 1L, 1L, Collections.<String>emptyList(), null, FileSource.APPEND,
+                Collections.<String>emptyList(), null, null, Collections.<String>emptyList());
+        return DataSplit.builder()
+                .rawConvertible(true)
+                .withPartition(BinaryRow.singleColumn(1))
+                .withBucket(1)
+                .withBucketPath("file://b1")
+                .withDataFiles(Collections.singletonList(dataFileMeta))
+                .build();
+    }
 }

fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/PaimonJdbcMetaStorePropertiesTest.java

@@ -174,4 +174,19 @@ public void testGetBackendPaimonOptions() throws Exception {
         Assertions.assertEquals("org.postgresql.Driver", backendOptions.get("jdbc.driver_class"));
         Assertions.assertEquals(2, backendOptions.size());
     }
+
+    @Test
+    public void testGetBackendPaimonOptionsRequiresDriverClass() throws Exception {
+        Map<String, String> props = new HashMap<>();
+        props.put("type", "paimon");
+        props.put("paimon.catalog.type", "jdbc");
+        props.put("uri", "jdbc:postgresql://127.0.0.1:5442/postgres");
+        props.put("warehouse", "s3://warehouse/path");
+        props.put("paimon.jdbc.driver_url", "file:///tmp/postgresql-42.5.0.jar");
+
+        PaimonJdbcMetaStoreProperties jdbcProps = (PaimonJdbcMetaStoreProperties) MetastoreProperties.create(props);
+        IllegalArgumentException exception = Assertions.assertThrows(IllegalArgumentException.class,
+                jdbcProps::getBackendPaimonOptions);
+        Assertions.assertTrue(exception.getMessage().contains("driver_class"));
+    }
 }

gensrc/thrift/PlanNodes.thrift

@@ -491,6 +491,9 @@ struct TFileScanRangeParams {
     // enable mapping varbinary type for Doris external table and TVF
     28: optional bool enable_mapping_varbinary = false;
     29: optional bool enable_mapping_timestamp_tz = false;
+    // Paimon options from FE, used for jni/native scanner
+    // Set at ScanNode level to avoid redundant serialization in each split
+    30: optional map<string, string> paimon_options
 }
 
 struct TFileRangeDesc {