[fix](fe) Fix Paimon JDBC driver registration for JNI scans

be/src/format/table/paimon_cpp_reader.cpp

@@ -269,8 +269,12 @@ 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 +314,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/java-common/src/main/java/org/apache/doris/common/classloader/JniScannerClassLoader.java

@@ -30,6 +30,15 @@ public JniScannerClassLoader(String scannerName, List<URL> urls, ClassLoader par
         this.scannerName = scannerName;
     }
 
+    public synchronized void addURLIfAbsent(URL url) {
+        for (URL existingUrl : getURLs()) {
+            if (existingUrl.equals(url)) {
+                return;
+            }
+        }
+        super.addURL(url);
+    }
+
     @Override
     public String toString() {
         return "JniScannerClassLoader{"

fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jdbc/JdbcDriverUtils.java

@@ -0,0 +1,114 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.common.jdbc;
+
+import org.apache.doris.common.classloader.JniScannerClassLoader;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.sql.Connection;
+import java.sql.Driver;
+import java.sql.DriverManager;
+import java.sql.DriverPropertyInfo;
+import java.sql.SQLFeatureNotSupportedException;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+public final class JdbcDriverUtils {
+    private static final ConcurrentHashMap<URL, ClassLoader> DRIVER_CLASS_LOADER_CACHE = new ConcurrentHashMap<>();
+    private static final Set<String> REGISTERED_DRIVER_KEYS = ConcurrentHashMap.newKeySet();
+
+    private JdbcDriverUtils() {
+    }
+
+    public static void registerDriver(String driverUrl, String driverClassName, ClassLoader classLoader) {
+        try {
+            URL url = new URL(driverUrl);
+            String driverKey = driverUrl + "#" + driverClassName;
+            if (!REGISTERED_DRIVER_KEYS.add(driverKey)) {
+                return;
+            }
+            try {
+                ClassLoader driverClassLoader = prepareDriverClassLoader(url, classLoader);
+                Class<?> loadedDriverClass = Class.forName(driverClassName, true, driverClassLoader);
+                Driver driver = (Driver) loadedDriverClass.getDeclaredConstructor().newInstance();
+                DriverManager.registerDriver(new DriverShim(driver));
+            } catch (Exception e) {
+                REGISTERED_DRIVER_KEYS.remove(driverKey);
+                throw new RuntimeException("Failed to register JDBC driver: " + driverClassName, e);
+            }
+        } catch (MalformedURLException e) {
+            throw new IllegalArgumentException("Invalid JDBC driver URL: " + driverUrl, e);
+        }
+    }
+
+    private static ClassLoader prepareDriverClassLoader(URL driverUrl, ClassLoader classLoader) {
+        if (classLoader instanceof JniScannerClassLoader) {
+            JniScannerClassLoader scannerClassLoader = (JniScannerClassLoader) classLoader;
+            scannerClassLoader.addURLIfAbsent(driverUrl);
+            return scannerClassLoader;
+        }
+        return DRIVER_CLASS_LOADER_CACHE.computeIfAbsent(driverUrl,
+                url -> URLClassLoader.newInstance(new URL[] {url}, classLoader));
+    }
+
+    private static final class DriverShim implements Driver {
+        private final Driver delegate;
+
+        private DriverShim(Driver delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public Connection connect(String url, java.util.Properties info) throws java.sql.SQLException {
+            return delegate.connect(url, info);
+        }
+
+        @Override
+        public boolean acceptsURL(String url) throws java.sql.SQLException {
+            return delegate.acceptsURL(url);
+        }
+
+        @Override
+        public DriverPropertyInfo[] getPropertyInfo(String url, java.util.Properties info)
+                throws java.sql.SQLException {
+            return delegate.getPropertyInfo(url, info);
+        }
+
+        @Override
+        public int getMajorVersion() {
+            return delegate.getMajorVersion();
+        }
+
+        @Override
+        public int getMinorVersion() {
+            return delegate.getMinorVersion();
+        }
+
+        @Override
+        public boolean jdbcCompliant() {
+            return delegate.jdbcCompliant();
+        }
+
+        @Override
+        public java.util.logging.Logger getParentLogger() throws SQLFeatureNotSupportedException {
+            return delegate.getParentLogger();
+        }
+    }
+}

fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJdbcDriverUtils.java

@@ -0,0 +1,59 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.paimon;
+
+import org.apache.doris.common.jdbc.JdbcDriverUtils;
+
+import java.util.Map;
+
+final class PaimonJdbcDriverUtils {
+    static final String PAIMON_JDBC_DRIVER_URL = "paimon.jdbc.driver_url";
+    static final String PAIMON_JDBC_DRIVER_CLASS = "paimon.jdbc.driver_class";
+    static final String JDBC_DRIVER_URL = "jdbc.driver_url";
+    static final String JDBC_DRIVER_CLASS = "jdbc.driver_class";
+
+    private PaimonJdbcDriverUtils() {
+    }
+
+    static void registerDriverIfNeeded(Map<String, String> params, ClassLoader parentClassLoader) {
+        String driverUrl = firstNonBlank(params.get(PAIMON_JDBC_DRIVER_URL), params.get(JDBC_DRIVER_URL));
+        if (driverUrl == null) {
+            return;
+        }
+        String driverClassName = firstNonBlank(params.get(PAIMON_JDBC_DRIVER_CLASS), params.get(JDBC_DRIVER_CLASS));
+        if (driverClassName == null) {
+            throw new IllegalArgumentException("paimon.jdbc.driver_class or jdbc.driver_class is required when "
+                    + "paimon.jdbc.driver_url or jdbc.driver_url is specified");
+        }
+        registerDriver(driverUrl, driverClassName, parentClassLoader);
+    }
+
+    static void registerDriver(String driverUrl, String driverClassName, ClassLoader parentClassLoader) {
+        JdbcDriverUtils.registerDriver(driverUrl, driverClassName, parentClassLoader);
+    }
+
+    private static String firstNonBlank(String first, String second) {
+        if (first != null && !first.trim().isEmpty()) {
+            return first;
+        }
+        if (second != null && !second.trim().isEmpty()) {
+            return second;
+        }
+        return null;
+    }
+}

fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java

@@ -92,6 +92,7 @@ public void open() throws IOException {
             // so we need to provide a classloader, otherwise it will cause NPE.
             Thread.currentThread().setContextClassLoader(classLoader);
             preExecutionAuthenticator.execute(() -> {
+                PaimonJdbcDriverUtils.registerDriverIfNeeded(params, classLoader);
                 initTable();
                 initReader();
                 return null;
@@ -227,4 +228,3 @@ private void initTable() {
     }
 
 }
-

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

@@ -0,0 +1,139 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.paimon;
+
+import org.apache.doris.common.classloader.JniScannerClassLoader;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Driver;
+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;
+import java.util.Properties;
+import java.util.jar.JarEntry;
+import java.util.jar.JarOutputStream;
+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(), ClassLoader.getPlatformClassLoader());
+        PaimonJdbcDriverUtils.registerDriverIfNeeded(params, scannerClassLoader);
+
+        Driver driver = DriverManager.getDriver("jdbc:dummy:test");
+        registeredDrivers.add(driver);
+        Assert.assertTrue(driver.acceptsURL("jdbc:dummy:test"));
+    }
+
+    @Test
+    public void testRegisterDriverIfNeededRequiresDriverClass() {
+        Map<String, String> params = new HashMap<>();
+        params.put(PaimonJdbcDriverUtils.PAIMON_JDBC_DRIVER_URL, "file:///tmp/postgresql-42.5.0.jar");
+
+        IllegalArgumentException exception = Assert.assertThrows(IllegalArgumentException.class,
+                () -> PaimonJdbcDriverUtils.registerDriverIfNeeded(params, getClass().getClassLoader()));
+        Assert.assertTrue(exception.getMessage().contains("driver_class"));
+    }
+
+    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)) {
+            Assert.assertNotNull(inputStream);
+            jarOutputStream.putNextEntry(new JarEntry(resourceName));
+            byte[] buffer = new byte[4096];
+            int bytesRead;
+            while ((bytesRead = inputStream.read(buffer)) >= 0) {
+                jarOutputStream.write(buffer, 0, bytesRead);
+            }
+            jarOutputStream.closeEntry();
+        }
+        return jarPath;
+    }
+
+    public static class DummyJdbcDriver implements Driver {
+        @Override
+        public java.sql.Connection connect(String url, Properties info) {
+            return null;
+        }
+
+        @Override
+        public boolean acceptsURL(String url) {
+            return url != null && url.startsWith("jdbc:dummy:");
+        }
+
+        @Override
+        public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) {
+            return new DriverPropertyInfo[0];
+        }
+
+        @Override
+        public int getMajorVersion() {
+            return 1;
+        }
+
+        @Override
+        public int getMinorVersion() {
+            return 0;
+        }
+
+        @Override
+        public boolean jdbcCompliant() {
+            return false;
+        }
+
+        @Override
+        public Logger getParentLogger() throws SQLFeatureNotSupportedException {
+            throw new SQLFeatureNotSupportedException("not supported");
+        }
+    }
+}