BugFix: Fix unable to connect to oracle tenant.

docs/spark-connector-oceanbase.md

@@ -103,8 +103,7 @@ mvn clean package -Dscala.version=2.11.12 -Dscala.binary.version=2.11 -DskipTest
 CREATE TEMPORARY VIEW spark_oceanbase
 USING oceanbase
 OPTIONS(
-  "host"= "localhost",
-  "sql-port" = "2881",
+  "url"= "jdbc:mysql://localhost:2881/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false",
   "schema-name"="test",
   "table-name"="test",
   "username"="root",
@@ -118,8 +117,7 @@ SELECT * FROM spark_oceanbase;
 
 ```scala
 val oceanBaseSparkDF = spark.read.format("OceanBase")
-  .option("host", "localhost")
-  .option("sql-port", 2881)
+  .option("url", "jdbc:mysql://localhost:2881/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false")
   .option("username", "root")
   .option("password", "123456")
   .option("table-name", "test")
@@ -178,8 +176,7 @@ CREATE TABLE test.orders (
 CREATE TEMPORARY VIEW test_jdbc
 USING oceanbase
 OPTIONS(
-  "host"="localhost",
-  "sql-port" = "2881",
+  "url"= "jdbc:mysql://localhost:2881/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false",
   "schema-name"="test",
   "table-name"="orders",
   "username"="root@test",
@@ -202,8 +199,7 @@ import org.apache.spark.sql.SaveMode
 df.write
   .format("oceanbase")
   .mode(saveMode = SaveMode.Append)
-  .option("host", "localhost")
-  .option("sql-port", 2881)
+  .option("url", "jdbc:mysql://localhost:2881/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false")
   .option("username", "root")
   .option("password", "123456")
   .option("table-name", "orders")
@@ -219,13 +215,13 @@ df.write
 CREATE TEMPORARY VIEW test_direct
 USING oceanbase
 OPTIONS(
-  "host"="localhost",
-  "sql-port" = "2881",
+  "url"="jdbc:mysql://localhost:2881/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false",
   "schema-name"="test",
   "table-name"="orders",
   "username"="root@test",
-  "password"="",
+  "password"="123456",
   "direct-load.enabled" = true,
+  "direct-load.host" = "localhost",
   "direct-load.rpc-port" = "2882"
 );
 
@@ -245,13 +241,13 @@ import org.apache.spark.sql.SaveMode
 df.write
   .format("oceanbase")
   .mode(saveMode = SaveMode.Append)
-  .option("host", "localhost")
-  .option("sql-port", 2881)
+  .option("url", "jdbc:mysql://localhost:2881/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false")
   .option("username", "root")
   .option("password", "123456")
   .option("table-name", "orders")
   .option("schema-name", "test")
   .option("direct-load.enabled", "true")
+  .option("direct-load.host", "localhost")
   .option("direct-load.rpc-port", "2882")
   .save()
 ```
@@ -272,16 +268,10 @@ df.write
         </thead>
         <tbody>
             <tr>
-                <td>host</td>
+                <td>url</td>
                 <td style="word-wrap: break-word;"></td>
                 <td>String</td>
-                <td>Hostname used in direct-load.</td>
-            </tr>
-            <tr>
-                <td>sql-port</td>
-                <td style="word-wrap: break-word;"></td>
-                <td>Integer</td>
-                <td>The SQL port.</td>
+                <td>The connection URL.</td>
             </tr>
             <tr>
                 <td>username</td>
@@ -336,6 +326,12 @@ df.write
                 <td>Boolean</td>
                 <td>Enable direct-load writing.</td>
             </tr>
+            <tr>
+                <td>direct-load.host</td>
+                <td></td>
+                <td>String</td>
+                <td>Hostname used in direct-load.</td>
+            </tr>
             <tr>
                 <td>direct-load.rpc-port</td>
                 <td>2882</td>

docs/spark-connector-oceanbase_cn.md

@@ -102,8 +102,7 @@ mvn clean package -Dscala.version=2.11.12 -Dscala.binary.version=2.11 -DskipTest
 CREATE TEMPORARY VIEW spark_oceanbase
 USING oceanbase
 OPTIONS(
-  "host"= "localhost",
-  "sql-port" = "2881",
+  "url"= "jdbc:mysql://localhost:2881/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false",
   "schema-name"="test",
   "table-name"="test",
   "username"="root",
@@ -117,8 +116,7 @@ SELECT * FROM spark_oceanbase;
 
 ```scala
 val oceanBaseSparkDF = spark.read.format("OceanBase")
-  .option("host", "localhost")
-  .option("sql-port", 2881)
+  .option("url", "jdbc:mysql://localhost:2881/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false")
   .option("username", "root")
   .option("password", "123456")
   .option("table-name", "test")
@@ -177,12 +175,11 @@ CREATE TABLE test.orders (
 CREATE TEMPORARY VIEW test_jdbc
 USING oceanbase
 OPTIONS(
-  "host"="localhost",
-  "sql-port" = "2881",
+  "url"="jdbc:mysql://localhost:2881/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false",
   "schema-name"="test",
   "table-name"="orders",
   "username"="root@test",
-  "password"=""
+  "password"="123456"
 );
 
 insert into table test_jdbc
@@ -203,8 +200,7 @@ import org.apache.spark.sql.SaveMode
 df.write
   .format("oceanbase")
   .mode(saveMode = SaveMode.Append)
-  .option("host", "localhost")
-  .option("sql-port", 2881)
+  .option("url", "jdbc:mysql://localhost:2881/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false")
   .option("username", "root")
   .option("password", "123456")
   .option("table-name", "orders")
@@ -220,13 +216,13 @@ df.write
 CREATE TEMPORARY VIEW test_direct
 USING oceanbase
 OPTIONS(
-  "host"="localhost",
-  "sql-port" = "2881",
+  "url"="jdbc:mysql://localhost:2881/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false",
   "schema-name"="test",
   "table-name"="orders",
   "username"="root@test",
-  "password"="",
+  "password"="123456",
   "direct-load.enabled" = true,
+  "direct-load.host" = "localhost",
   "direct-load.rpc-port" = "2882"
 );
 
@@ -248,13 +244,13 @@ import org.apache.spark.sql.SaveMode
 df.write
   .format("oceanbase")
   .mode(saveMode = SaveMode.Append)
-  .option("host", "localhost")
-  .option("sql-port", 2881)
+  .option("url", "jdbc:mysql://localhost:2881/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false")
   .option("username", "root")
   .option("password", "123456")
   .option("table-name", "orders")
   .option("schema-name", "test")
   .option("direct-load.enabled", "true")
+  .option("direct-load.host", "localhost")
   .option("direct-load.rpc-port", "2882")
   .save()
 ```
@@ -275,16 +271,10 @@ df.write
         </thead>
         <tbody>
             <tr>
-                <td>host</td>
+                <td>url</td>
                 <td style="word-wrap: break-word;"></td>
                 <td>String</td>
-                <td>数据库的 JDBC url。</td>
-            </tr>
-            <tr>
-                <td>sql-port</td>
-                <td style="word-wrap: break-word;">2881</td>
-                <td>Int</td>
-                <td>SQL 端口。</td>
+                <td>连接到OceanBase的 JDBC url.</td>
             </tr>
             <tr>
                 <td>username</td>
@@ -333,6 +323,12 @@ df.write
                 <td>Boolean</td>
                 <td>是否开启旁路导入写入。</td>
             </tr>
+            <tr>
+                <td>direct-load.host</td>
+                <td></td>
+                <td>String</td>
+                <td>旁路导入用到的host地址。</td>
+            </tr>
             <tr>
                 <td>direct-load.rpc-port</td>
                 <td>2882</td>

spark-connector-oceanbase-common/src/main/java/com/oceanbase/spark/config/ConfigBuilder.java

@@ -15,6 +15,7 @@
  */
 package com.oceanbase.spark.config;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -191,4 +192,29 @@ public ConfigEntry<Boolean> booleanConf() {
 
         return conf;
     }
+
+    /**
+     * Creates a configuration entry for Duration data type.
+     *
+     * @return The created ConfigEntry instance for Boolean data type.
+     */
+    public ConfigEntry<Duration> durationConf() {
+        ConfigEntry<Duration> conf =
+                new ConfigEntry<>(key, version, doc, alternatives, isPublic, isDeprecated);
+        Function<String, Duration> func =
+                s -> {
+                    if (s == null || s.isEmpty()) {
+                        return null;
+                    } else {
+                        return Duration.parse(s);
+                    }
+                };
+        conf.setValueConverter(func);
+
+        Function<Duration, String> stringFunc =
+                t -> Optional.ofNullable(t).map(String::valueOf).orElse(null);
+        conf.setStringConverter(stringFunc);
+
+        return conf;
+    }
 }

spark-connector-oceanbase-e2e-tests/src/test/java/com/oceanbase/spark/OceanBaseE2eITCase.java

@@ -80,20 +80,20 @@ public void testInsertValues() throws Exception {
                         "CREATE TEMPORARY VIEW test_sink "
                                 + "USING oceanbase "
                                 + "OPTIONS( "
-                                + "  \"host\"= \"%s\","
-                                + "  \"sql-port\" = \"%s\","
+                                + "  \"url\"= \"%s\","
                                 + "  \"schema-name\"=\"%s\","
                                 + "  \"table-name\"=\"products\","
                                 + "  \"username\"=\"%s\","
                                 + "  \"password\"=\"%s\","
                                 + "  \"direct-load.enabled\" = \"true\","
+                                + "  \"direct-load.host\" = \"%s\","
                                 + "  \"direct-load.rpc-port\" = \"%s\" "
                                 + ");",
-                        getHostInContainer(),
-                        getPortInContainer(),
+                        getJdbcUrlInContainer(),
                         getSchemaName(),
                         getUsername(),
                         getPassword(),
+                        getHostInContainer(),
                         getRpcPortInContainer()));
 
         sqlLines.add(
@@ -163,20 +163,20 @@ public void testInsertValuesSpark2() throws Exception {
                                 + "df.write\n"
                                 + "  .format(\"oceanbase\")\n"
                                 + "  .mode(saveMode = SaveMode.Append)\n"
-                                + "  .option(\"host\", \"%s\")\n"
-                                + "  .option(\"sql-port\", \"%s\")\n"
+                                + "  .option(\"url\", \"%s\")\n"
                                 + "  .option(\"username\", \"%s\")\n"
                                 + "  .option(\"password\", \"%s\")\n"
                                 + "  .option(\"table-name\", \"products\")\n"
                                 + "  .option(\"schema-name\", \"%s\")\n"
                                 + "  .option(\"direct-load.enabled\", value = true)\n"
+                                + "  .option(\"direct-load.host\", value = \"%s\")\n"
                                 + "  .option(\"direct-load.rpc-port\", value = \"%s\")\n"
                                 + "  .save()",
-                        getHostInContainer(),
-                        getPortInContainer(),
+                        getJdbcUrlInContainer(),
                         getUsername(),
                         getPassword(),
                         getSchemaName(),
+                        getHostInContainer(),
                         getRpcPortInContainer()));
 
         submitSparkShellJob(

spark-connector-oceanbase-e2e-tests/src/test/java/com/oceanbase/spark/utils/SparkContainerTestEnvironment.java

@@ -82,6 +82,16 @@ public void after() throws Exception {
         }
     }
 
+    public String getJdbcUrlInContainer() {
+        return "jdbc:mysql://"
+                + getHostInContainer()
+                + ":"
+                + getPortInContainer()
+                + "/"
+                + getSchemaName()
+                + "?useUnicode=true&characterEncoding=UTF-8&useSSL=false";
+    }
+
     public String getHostInContainer() {
         return getOBServerIP();
     }

spark-connector-oceanbase/pom.xml

@@ -37,6 +37,12 @@ under the License.
     </modules>
 
     <dependencies>
+        <dependency>
+            <groupId>com.oceanbase</groupId>
+            <artifactId>spark-connector-oceanbase-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>com.oceanbase</groupId>
             <artifactId>obkv-table-client</artifactId>

spark-connector-oceanbase/spark-connector-oceanbase-2.4/pom.xml

@@ -90,6 +90,7 @@ under the License.
                             <artifactSet>
                                 <includes combine.children="append">
                                     <include>com.alibaba:*</include>
+                                    <include>com.oceanbase:spark-connector-oceanbase-common</include>
                                     <include>com.oceanbase:spark-connector-oceanbase-base</include>
                                     <include>com.oceanbase:obkv-table-client</include>
                                     <include>com.alipay.sofa:bolt</include>

spark-connector-oceanbase/spark-connector-oceanbase-3.1/pom.xml

@@ -69,6 +69,7 @@ under the License.
                             <artifactSet>
                                 <includes combine.children="append">
                                     <include>com.alibaba:*</include>
+                                    <include>com.oceanbase:spark-connector-oceanbase-common</include>
                                     <include>com.oceanbase:spark-connector-oceanbase-base</include>
                                     <include>com.oceanbase:obkv-table-client</include>
                                     <include>com.alipay.sofa:bolt</include>

spark-connector-oceanbase/spark-connector-oceanbase-3.2/pom.xml

@@ -69,6 +69,7 @@ under the License.
                             <artifactSet>
                                 <includes combine.children="append">
                                     <include>com.alibaba:*</include>
+                                    <include>com.oceanbase:spark-connector-oceanbase-common</include>
                                     <include>com.oceanbase:spark-connector-oceanbase-base</include>
                                     <include>com.oceanbase:obkv-table-client</include>
                                     <include>com.alipay.sofa:bolt</include>

spark-connector-oceanbase/spark-connector-oceanbase-3.3/pom.xml

@@ -69,6 +69,7 @@ under the License.
                             <artifactSet>
                                 <includes combine.children="append">
                                     <include>com.alibaba:*</include>
+                                    <include>com.oceanbase:spark-connector-oceanbase-common</include>
                                     <include>com.oceanbase:spark-connector-oceanbase-base</include>
                                     <include>com.oceanbase:obkv-table-client</include>
                                     <include>com.alipay.sofa:bolt</include>

spark-connector-oceanbase/spark-connector-oceanbase-3.4/pom.xml

@@ -69,6 +69,7 @@ under the License.
                             <artifactSet>
                                 <includes combine.children="append">
                                     <include>com.alibaba:*</include>
+                                    <include>com.oceanbase:spark-connector-oceanbase-common</include>
                                     <include>com.oceanbase:spark-connector-oceanbase-base</include>
                                     <include>com.oceanbase:obkv-table-client</include>
                                     <include>com.alipay.sofa:bolt</include>

spark-connector-oceanbase/spark-connector-oceanbase-base/pom.xml

@@ -169,6 +169,11 @@ under the License.
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>com.oceanbase</groupId>
+            <artifactId>oceanbase-client</artifactId>
+            <scope>test</scope>
+        </dependency>
 
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>