This is an automated email from the ASF dual-hosted git repository. CalvinKirs pushed a commit to branch revert-66596-cloud_download in repository https://gitbox.apache.org/repos/asf/doris.git
commit f41ed52dc41a6cd2ccc2678fcb12cc01e89d1a00 Author: Calvin Kirs <[email protected]> AuthorDate: Tue Aug 25 15:56:41 2026 +0800 Revert "[refactor](plugin) Remove FE cloud auto-download for JDBC drivers and…" This reverts commit 994630bb2ef85aa70487e4004c807991e85b4ce2. --- .../org/apache/doris/catalog/JdbcResource.java | 14 ++ .../doris/common/plugin/CloudPluginDownloader.java | 157 +++++++++++++++++++ .../plans/commands/CreateFunctionCommand.java | 16 ++ .../common/plugin/CloudPluginDownloaderTest.java | 173 +++++++++++++++++++++ .../test_cloud_plugin_auto_download.groovy | 121 ++++++++++++++ 5 files changed, 481 insertions(+) diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java index 43ecf7d5cd8..0141d33d276 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java @@ -23,6 +23,8 @@ import org.apache.doris.common.Config; import org.apache.doris.common.DdlException; import org.apache.doris.common.EnvUtils; import org.apache.doris.common.FeConstants; +import org.apache.doris.common.plugin.CloudPluginDownloader; +import org.apache.doris.common.plugin.CloudPluginDownloader.PluginType; import org.apache.doris.common.proc.BaseProcResult; import org.apache.doris.common.util.TimeUtils; import org.apache.doris.common.util.Util; @@ -447,6 +449,18 @@ public class JdbcResource extends Resource { } else if (oldTargetFile.exists()) { // File exists in old default directory return "file://" + oldTargetPath; + } else if (Config.isCloudMode()) { + // Cloud mode: download from cloud to default directory + try { + String downloadedPath = CloudPluginDownloader.downloadFromCloud( + PluginType.JDBC_DRIVERS, driverUrl, targetPath); + return "file://" + downloadedPath; + } catch (Exception e) { + LOG.warn("failed to download jdbc driver url: " + driverUrl, e); + throw new RuntimeException("Cannot download JDBC driver from cloud: " + driverUrl + + ". Please retry later or check your driver has been uploaded to cloud. Error: " + + Util.getRootCauseMessage(e)); + } } else { // File does not exist in both new and old default directory throw new RuntimeException("JDBC driver file does not exist: " + driverUrl); diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/plugin/CloudPluginDownloader.java b/fe/fe-core/src/main/java/org/apache/doris/common/plugin/CloudPluginDownloader.java new file mode 100644 index 00000000000..20e6c0342d3 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/common/plugin/CloudPluginDownloader.java @@ -0,0 +1,157 @@ +// 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.plugin; + +import org.apache.doris.cloud.proto.Cloud; +import org.apache.doris.cloud.rpc.MetaServiceProxy; +import org.apache.doris.filesystem.DorisInputFile; +import org.apache.doris.filesystem.Location; +import org.apache.doris.fs.FileSystemFactory; +import org.apache.doris.service.FrontendOptions; + +import com.google.common.base.Strings; + +import java.io.File; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.StandardCopyOption; +import java.util.HashMap; +import java.util.Map; + +/** + * Simple cloud plugin downloader for UDF and JDBC drivers. + */ +public class CloudPluginDownloader { + + public enum PluginType { + JDBC_DRIVERS, + JAVA_UDF, + CONNECTORS, // Reserved, not supported yet + HADOOP_CONF // Reserved, not supported yet + } + + /** + * Download plugin from cloud storage to local path + */ + public static synchronized String downloadFromCloud(PluginType type, String name, String localPath) { + validateInput(type, name); + try { + Cloud.ObjectStoreInfoPB objInfo = getCloudStorageInfo(); + String remotePath = buildS3Path(objInfo, type, name); + return doDownload(objInfo, remotePath, localPath); + } catch (Exception e) { + throw new RuntimeException("Failed to download plugin: " + e.getMessage(), e); + } + } + + /** + * Validate input parameters + */ + static void validateInput(PluginType type, String name) { + if (Strings.isNullOrEmpty(name)) { + throw new IllegalArgumentException("Plugin name cannot be empty"); + } + + if (type != PluginType.JDBC_DRIVERS && type != PluginType.JAVA_UDF) { + throw new UnsupportedOperationException("Plugin type " + type + " is not supported yet"); + } + } + + /** + * Get cloud storage info from MetaService + * Package-private for testing + */ + static Cloud.ObjectStoreInfoPB getCloudStorageInfo() throws Exception { + Cloud.GetObjStoreInfoResponse response = MetaServiceProxy.getInstance() + .getObjStoreInfo(Cloud.GetObjStoreInfoRequest.newBuilder() + .setRequestIp(FrontendOptions.getLocalHostAddressCached()) + .build()); + + if (response.getStatus().getCode() != Cloud.MetaServiceCode.OK) { + throw new RuntimeException("Failed to get storage info: " + response.getStatus().getMsg()); + } + + if (response.getObjInfoList().isEmpty()) { + throw new RuntimeException("Only SaaS cloud storage is supported currently"); + } + + return response.getObjInfo(0); + } + + /** + * Build complete S3 path from objInfo + * Package-private for testing + */ + static String buildS3Path(Cloud.ObjectStoreInfoPB objInfo, PluginType type, String name) { + String bucket = objInfo.getBucket(); + String prefix = objInfo.hasPrefix() ? objInfo.getPrefix() : ""; + String relativePath = String.format("plugins/%s/%s", type.name().toLowerCase(), name); + + String fullPath; + if (Strings.isNullOrEmpty(prefix)) { + fullPath = bucket + "/" + relativePath; + } else { + fullPath = bucket + "/" + prefix + "/" + relativePath; + } + + return "s3://" + fullPath; + } + + /** + * Execute download using SPI FileSystem + */ + private static String doDownload(Cloud.ObjectStoreInfoPB objInfo, String remotePath, String localPath) + throws Exception { + // Create parent directory + Path parentDir = Paths.get(localPath).getParent(); + if (parentDir != null && !Files.exists(parentDir)) { + Files.createDirectories(parentDir); + } + + // Delete existing file if present + File localFile = new File(localPath); + if (localFile.exists() && !localFile.delete()) { + throw new RuntimeException("Failed to delete existing file: " + localPath); + } + + // Download via SPI FileSystem + Map<String, String> properties = buildProperties(objInfo); + org.apache.doris.filesystem.FileSystem fileSystem = + FileSystemFactory.getFileSystem(properties); + DorisInputFile inputFile = fileSystem.newInputFile(Location.of(remotePath)); + try (InputStream in = inputFile.newStream()) { + Files.copy(in, localFile.toPath(), StandardCopyOption.REPLACE_EXISTING); + } + return localPath; + } + + /** + * Build storage properties map from objInfo + */ + private static Map<String, String> buildProperties(Cloud.ObjectStoreInfoPB objInfo) { + Map<String, String> props = new HashMap<>(); + props.put("s3.endpoint", objInfo.getEndpoint()); + props.put("s3.region", objInfo.getRegion()); + props.put("s3.access_key", objInfo.getAk()); + props.put("s3.secret_key", objInfo.getSk()); + props.put("s3.bucket", objInfo.getBucket()); + return props; + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateFunctionCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateFunctionCommand.java index 0cdb1be3c5d..66a4c1f428c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateFunctionCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateFunctionCommand.java @@ -46,6 +46,7 @@ import org.apache.doris.common.EnvUtils; import org.apache.doris.common.ErrorCode; import org.apache.doris.common.ErrorReport; import org.apache.doris.common.FeConstants; +import org.apache.doris.common.plugin.CloudPluginDownloader; import org.apache.doris.common.util.URI; import org.apache.doris.common.util.Util; import org.apache.doris.mysql.privilege.PrivPredicate; @@ -484,6 +485,21 @@ public class CreateFunctionCommand extends Command implements ForwardWithSync { private String checkAndReturnDefaultJavaUdfUrl(String url) { String defaultUrl = EnvUtils.getDorisHome() + "/plugins/java_udf"; + // In cloud mode, try cloud download first + if (Config.isCloudMode()) { + String targetPath = defaultUrl + "/" + url; + try { + String downloadedPath = CloudPluginDownloader.downloadFromCloud( + CloudPluginDownloader.PluginType.JAVA_UDF, url, targetPath); + if (!downloadedPath.isEmpty()) { + return "file://" + downloadedPath; + } + } catch (Exception e) { + throw new RuntimeException("Cannot download UDF from cloud: " + url + + ". Please retry later or check your UDF has been uploaded to cloud."); + } + } + // Return the file path (original UDF behavior) return "file://" + defaultUrl + "/" + url; } diff --git a/fe/fe-core/src/test/java/org/apache/doris/common/plugin/CloudPluginDownloaderTest.java b/fe/fe-core/src/test/java/org/apache/doris/common/plugin/CloudPluginDownloaderTest.java new file mode 100644 index 00000000000..50b91fc786b --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/common/plugin/CloudPluginDownloaderTest.java @@ -0,0 +1,173 @@ +// 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.plugin; + +import org.apache.doris.cloud.proto.Cloud; +import org.apache.doris.cloud.rpc.MetaServiceProxy; +import org.apache.doris.common.plugin.CloudPluginDownloader.PluginType; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.Collections; + +/** + * Unit tests for CloudPluginDownloader using package-private methods for direct white-box testing. + */ +public class CloudPluginDownloaderTest { + + private Cloud.GetObjStoreInfoResponse mockResponse; + private Cloud.ObjectStoreInfoPB mockObjInfo; + private MetaServiceProxy mockMetaServiceProxy; + + @BeforeEach + void setUp() { + mockResponse = Mockito.mock(Cloud.GetObjStoreInfoResponse.class); + mockObjInfo = Mockito.mock(Cloud.ObjectStoreInfoPB.class); + mockMetaServiceProxy = Mockito.mock(MetaServiceProxy.class); + } + + // ============== validateInput Tests ============== + + @Test + void testValidateInput() { + // Positive cases + Assertions.assertDoesNotThrow(() -> { + CloudPluginDownloader.validateInput(PluginType.JDBC_DRIVERS, "mysql.jar"); + CloudPluginDownloader.validateInput(PluginType.JAVA_UDF, "my_udf.jar"); + }); + + // Empty/null name + IllegalArgumentException ex1 = Assertions.assertThrows(IllegalArgumentException.class, + () -> CloudPluginDownloader.validateInput(PluginType.JDBC_DRIVERS, "")); + Assertions.assertEquals("Plugin name cannot be empty", ex1.getMessage()); + + IllegalArgumentException ex2 = Assertions.assertThrows(IllegalArgumentException.class, + () -> CloudPluginDownloader.validateInput(PluginType.JDBC_DRIVERS, null)); + Assertions.assertEquals("Plugin name cannot be empty", ex2.getMessage()); + + // Unsupported types + UnsupportedOperationException ex3 = Assertions.assertThrows(UnsupportedOperationException.class, + () -> CloudPluginDownloader.validateInput(PluginType.CONNECTORS, "test.jar")); + Assertions.assertTrue(ex3.getMessage().contains("is not supported yet")); + } + + // ============== getCloudStorageInfo Tests ============== + + @Test + void testGetCloudStorageInfo() throws Exception { + try (MockedStatic<MetaServiceProxy> mockedStatic = Mockito.mockStatic(MetaServiceProxy.class)) { + mockedStatic.when(MetaServiceProxy::getInstance).thenReturn(mockMetaServiceProxy); + + // Success case + Cloud.MetaServiceResponseStatus okStatus = Cloud.MetaServiceResponseStatus.newBuilder() + .setCode(Cloud.MetaServiceCode.OK).build(); + Mockito.when(mockResponse.getStatus()).thenReturn(okStatus); + Mockito.when(mockResponse.getObjInfoList()).thenReturn(Collections.singletonList(mockObjInfo)); + Mockito.when(mockResponse.getObjInfo(0)).thenReturn(mockObjInfo); + Mockito.when(mockMetaServiceProxy.getObjStoreInfo(Mockito.any())).thenReturn(mockResponse); + + Cloud.ObjectStoreInfoPB result = CloudPluginDownloader.getCloudStorageInfo(); + Assertions.assertEquals(mockObjInfo, result); + + // Error response + Cloud.MetaServiceResponseStatus failedStatus = Cloud.MetaServiceResponseStatus.newBuilder() + .setCode(Cloud.MetaServiceCode.INVALID_ARGUMENT).setMsg("Test error").build(); + Mockito.when(mockResponse.getStatus()).thenReturn(failedStatus); + + RuntimeException ex1 = Assertions.assertThrows(RuntimeException.class, + CloudPluginDownloader::getCloudStorageInfo); + Assertions.assertTrue(ex1.getMessage().contains("Failed to get storage info")); + + // Empty storage list + Mockito.when(mockResponse.getStatus()).thenReturn(okStatus); + Mockito.when(mockResponse.getObjInfoList()).thenReturn(Collections.emptyList()); + + RuntimeException ex2 = Assertions.assertThrows(RuntimeException.class, + CloudPluginDownloader::getCloudStorageInfo); + Assertions.assertTrue(ex2.getMessage().contains("Only SaaS cloud storage is supported")); + } + } + + // ============== buildS3Path Tests ============== + + @Test + void testBuildS3Path() { + Mockito.when(mockObjInfo.getBucket()).thenReturn("test-bucket"); + + // With prefix + Mockito.when(mockObjInfo.hasPrefix()).thenReturn(true); + Mockito.when(mockObjInfo.getPrefix()).thenReturn("test-prefix"); + Assertions.assertEquals("s3://test-bucket/test-prefix/plugins/jdbc_drivers/mysql.jar", + CloudPluginDownloader.buildS3Path(mockObjInfo, PluginType.JDBC_DRIVERS, "mysql.jar")); + + // Without prefix + Mockito.when(mockObjInfo.hasPrefix()).thenReturn(false); + Assertions.assertEquals("s3://test-bucket/plugins/java_udf/my_udf.jar", + CloudPluginDownloader.buildS3Path(mockObjInfo, PluginType.JAVA_UDF, "my_udf.jar")); + + // All plugin types + Assertions.assertEquals("s3://test-bucket/plugins/connectors/test.jar", + CloudPluginDownloader.buildS3Path(mockObjInfo, PluginType.CONNECTORS, "test.jar")); + Assertions.assertEquals("s3://test-bucket/plugins/hadoop_conf/test.xml", + CloudPluginDownloader.buildS3Path(mockObjInfo, PluginType.HADOOP_CONF, "test.xml")); + } + + // ============== Integration Test ============== + + @Test + void testDownloadFromCloudIntegration() { + // Basic integration test - should fail early due to validation + IllegalArgumentException ex = Assertions.assertThrows(IllegalArgumentException.class, + () -> CloudPluginDownloader.downloadFromCloud(PluginType.JDBC_DRIVERS, "", "/tmp/test.jar")); + Assertions.assertEquals("Plugin name cannot be empty", ex.getMessage()); + + // Should fail at MetaService level (no real cloud environment) + RuntimeException ex2 = Assertions.assertThrows(RuntimeException.class, + () -> CloudPluginDownloader.downloadFromCloud(PluginType.JDBC_DRIVERS, "mysql.jar", "/tmp/test.jar")); + Assertions.assertTrue(ex2.getMessage().contains("Failed to download plugin")); + } + + @Test + void testBuildS3PathEdgeCases() { + // Test empty bucket (edge case) + Mockito.when(mockObjInfo.getBucket()).thenReturn(""); + Mockito.when(mockObjInfo.hasPrefix()).thenReturn(false); + String result = CloudPluginDownloader.buildS3Path(mockObjInfo, PluginType.JDBC_DRIVERS, "test.jar"); + Assertions.assertEquals("s3:///plugins/jdbc_drivers/test.jar", result); + + // Test special characters in name + Mockito.when(mockObjInfo.getBucket()).thenReturn("test-bucket"); + String specialResult = CloudPluginDownloader.buildS3Path(mockObjInfo, PluginType.JAVA_UDF, "[email protected]"); + Assertions.assertEquals("s3://test-bucket/plugins/java_udf/[email protected]", specialResult); + } + + // ============== Enum Tests ============== + + @Test + void testPluginTypeEnum() { + Assertions.assertEquals("JDBC_DRIVERS", PluginType.JDBC_DRIVERS.name()); + Assertions.assertEquals("JAVA_UDF", PluginType.JAVA_UDF.name()); + Assertions.assertEquals("CONNECTORS", PluginType.CONNECTORS.name()); + Assertions.assertEquals("HADOOP_CONF", PluginType.HADOOP_CONF.name()); + Assertions.assertEquals(4, PluginType.values().length); + } +} diff --git a/regression-test/suites/plugin_p1/test_cloud_plugin_auto_download.groovy b/regression-test/suites/plugin_p1/test_cloud_plugin_auto_download.groovy new file mode 100644 index 00000000000..2b697db07ba --- /dev/null +++ b/regression-test/suites/plugin_p1/test_cloud_plugin_auto_download.groovy @@ -0,0 +1,121 @@ +// 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. + +suite("test_cloud_plugin_auto_download", "p1,external") { + + //sass cloud-mode only + if (!isCloudMode() || enableStoragevault()) { + logger.info("Skip test_plugin_auto_download because not in sass cloud mode") + return + } + + String jdbcUrl = context.config.jdbcUrl + String jdbcUser = context.config.jdbcUser + String jdbcPassword = context.config.jdbcPassword + + sql """drop database if exists internal.test_auto_download_db; """ + sql """create database if not exists internal.test_auto_download_db;""" + sql """create table if not exists internal.test_auto_download_db.test_tbl + (id int, name varchar(20)) + distributed by hash(id) buckets 1 + properties('replication_num' = '1'); + """ + sql """insert into internal.test_auto_download_db.test_tbl values(1, 'auto_download_test')""" + + sql """drop catalog if exists test_auto_download_catalog """ + sql """ CREATE CATALOG `test_auto_download_catalog` PROPERTIES ( + "user" = "${jdbcUser}", + "type" = "jdbc", + "password" = "${jdbcPassword}", + "jdbc_url" = "${jdbcUrl}", + "driver_url" = "mysql-connector-j-8.3.0.jar", + "driver_class" = "com.mysql.cj.jdbc.Driver" + )""" + + def result = sql """ + select * from test_auto_download_catalog.test_auto_download_db.test_tbl + """ + logger.info("result: ${result}") + assertTrue(result.size() > 0) + assertEquals(result[0][0], 1) + assertEquals(result[0][1], "auto_download_test") + + sql """drop catalog if exists test_auto_download_catalog """ + + sql """ use internal.test_auto_download_db; """ + + sql """DROP FUNCTION IF EXISTS java_udf_add_one(int)""" + + sql """ CREATE FUNCTION java_udf_add_one(int) RETURNS int PROPERTIES ( + "file"="java-udf-demo-jar-with-dependencies.jar", + "symbol"="org.apache.doris.udf.AddOne", + "type"="JAVA_UDF" + ); """ + + def result2 = sql """ + select java_udf_add_one(100) as result + """ + assertTrue(result2.size() > 0) + assertEquals(result2[0][0], 101) + + sql """DROP FUNCTION IF EXISTS java_udf_add_one(int)""" + + // negative test case 1: non-existent JDBC driver jar + sql """drop catalog if exists test_non_existent_driver_catalog """ + try { + sql """ CREATE CATALOG `test_non_existent_driver_catalog` PROPERTIES ( + "user" = "${jdbcUser}", + "type" = "jdbc", + "password" = "${jdbcPassword}", + "jdbc_url" = "${jdbcUrl}", + "driver_url" = "non-existent-mysql-driver.jar", + "driver_class" = "com.mysql.cj.jdbc.Driver" + )""" + + sql """ + select * from test_non_existent_driver_catalog.test_auto_download_db.test_tbl + """ + assertTrue(false, "Should have thrown exception for non-existent driver jar") + } catch (Exception e) { + logger.info("Expected exception for non-existent driver jar: " + e.getMessage()) + assertTrue(e.getMessage().contains("has been uploaded to cloud")) + } finally { + sql """drop catalog if exists test_non_existent_driver_catalog """ + } + + // negative test case 2: non-existent UDF jar + sql """DROP FUNCTION IF EXISTS java_udf_non_existent(int)""" + try { + sql """ CREATE FUNCTION java_udf_non_existent(int) RETURNS int PROPERTIES ( + "file"="non-existent-udf.jar", + "symbol"="org.apache.doris.udf.NonExistent", + "type"="JAVA_UDF" + ); """ + + sql """ + select java_udf_non_existent(100) as result + """ + assertTrue(false, "Should have thrown exception for non-existent UDF jar") + } catch (Exception e) { + logger.info("Expected exception for non-existent UDF jar: " + e.getMessage()) + assertTrue(e.getMessage().contains("has been uploaded to cloud")) + } finally { + sql """DROP FUNCTION IF EXISTS java_udf_non_existent(int)""" + } + + sql """ drop database if exists internal.test_auto_download_db; """ +} \ No newline at end of file --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
