Enhancement: Add direct-load.username configuration option to be compatible with OceanBase on the cloud. (#38)

* Enhancement: Added direct-load.username configuration option to be compatible with OceanBase on the cloud.

Author: yuanoOo

docs/spark-catalog-oceanbase.md

@@ -371,6 +371,13 @@ select * from spark_catalog.default.orders;
                 <td>Integer</td>
                 <td>Rpc port used in direct-load.</td>
             </tr>
+            <tr>
+                <td>spark.sql.catalog.your_catalog_name.direct-load.username</td>
+                <td>否</td>
+                <td></td>
+                <td>String</td>
+                <td>The direct-load's username. If this configuration is not specified, the jdbc username is used.</td>
+            </tr>
             <tr>
                 <td>spark.sql.catalog.your_catalog_name.direct-load.parallel</td>
                 <td>No</td>

docs/spark-catalog-oceanbase_cn.md

@@ -370,6 +370,13 @@ select * from spark_catalog.default.orders;
                 <td>Integer</td>
                 <td>旁路导入用到的rpc端口。</td>
             </tr>
+            <tr>
+                <td>spark.sql.catalog.your_catalog_name.direct-load.username</td>
+                <td>否</td>
+                <td></td>
+                <td>String</td>
+                <td>旁路导入用户名。如果不指定该配置，则使用jdbc用户名。</td>
+            </tr>
             <tr>
                 <td>spark.sql.catalog.your_catalog_name.direct-load.parallel</td>
                 <td>否</td>

spark-connector-oceanbase/spark-connector-oceanbase-base/src/main/java/com/oceanbase/spark/config/OceanBaseConfig.java

@@ -88,6 +88,13 @@ public class OceanBaseConfig extends Config implements Serializable {
                     .checkValue(port -> port > 0, ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
                     .createWithDefault(2882);
 
+    public static final ConfigEntry<String> DIRECT_LOAD_USERNAME =
+            new ConfigBuilder("direct-load.username")
+                    .doc("The username used in direct-load")
+                    .version(ConfigConstants.VERSION_1_1_0)
+                    .stringConf()
+                    .create();
+
     public static final ConfigEntry<Integer> DIRECT_LOAD_PARALLEL =
             new ConfigBuilder("direct-load.parallel")
                     .doc(
@@ -159,7 +166,7 @@ public class OceanBaseConfig extends Config implements Serializable {
                     .createWithDefault(0);
 
     public static final ConfigEntry<Integer> DIRECT_LOAD_BATCH_SIZE =
-            new ConfigBuilder("direct-load-batch-size")
+            new ConfigBuilder("direct-load.batch-size")
                     .doc("The batch size write to OceanBase one time")
                     .version(ConfigConstants.VERSION_1_0_0)
                     .intConf()
@@ -289,6 +296,10 @@ public int getDirectLoadPort() {
         return get(DIRECT_LOAD_RPC_PORT);
     }
 
+    public String getDirectLoadUserName() {
+        return get(DIRECT_LOAD_USERNAME);
+    }
+
     public String getDirectLoadExecutionId() {
         return get(DIRECT_LOAD_EXECUTION_ID);
     }

spark-connector-oceanbase/spark-connector-oceanbase-base/src/main/java/com/oceanbase/spark/directload/DirectLoadUtils.java

@@ -25,10 +25,14 @@ public class DirectLoadUtils {
     public static DirectLoader buildDirectLoaderFromSetting(OceanBaseConfig oceanBaseConfig) {
         try {
             OceanBaseUserInfo userInfo = OceanBaseUserInfo.parse(oceanBaseConfig);
+            String directLoadUserName = oceanBaseConfig.getDirectLoadUserName();
+            if (directLoadUserName == null || directLoadUserName.isEmpty()) {
+                directLoadUserName = userInfo.getUser();
+            }
             return new DirectLoaderBuilder()
                     .host(oceanBaseConfig.getDirectLoadHost())
                     .port(oceanBaseConfig.getDirectLoadPort())
-                    .user(userInfo.getUser())
+                    .user(directLoadUserName)
                     .password(oceanBaseConfig.getPassword())
                     .tenant(userInfo.getTenant())
                     .schema(oceanBaseConfig.getSchemaName())

spark-connector-oceanbase/spark-connector-oceanbase-base/src/test/scala/com/oceanbase/spark/OBCatalogMySQLITCase.scala

@@ -153,6 +153,7 @@ class OBCatalogMySQLITCase extends OceanBaseMySQLTestBase {
       .config("spark.sql.catalog.ob.direct-load.enabled", "true")
       .config("spark.sql.catalog.ob.direct-load.host", getHost)
       .config("spark.sql.catalog.ob.direct-load.rpc-port", getRpcPort)
+      .config("spark.sql.catalog.ob.direct-load.username", getUsername.split("@").head)
       .getOrCreate()
 
     insertTestData(session, "products")

spark-connector-oceanbase/spark-connector-oceanbase-base/src/test/scala/com/oceanbase/spark/OceanBaseMySQLConnectorITCase.scala

@@ -75,6 +75,7 @@ class OceanBaseMySQLConnectorITCase extends OceanBaseMySQLTestBase {
                    |  "direct-load.enabled"=true,
                    |  "direct-load.host"="$getHost",
                    |  "direct-load.rpc-port"=$getRpcPort,
+                   |  "direct-load.username"="${getUsername.split("@").head}",
                    |  "direct-load.write-thread-num"="1"
                    |);
                    |""".stripMargin)