Merge pull request #9 from weiguoz/polish_code

Polish code

Author: weiguoz

pom.xml

@@ -40,12 +40,6 @@
             <artifactId>commons-lang3</artifactId>
             <version>3.9</version>
         </dependency>
-        <!-- coding tools -->
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-            <version>1.18.10</version>
-        </dependency>
         <!-- unit test tools -->
         <dependency>
             <groupId>junit</groupId>

src/main/java/org/sqlflow/client/SQLFlow.java

@@ -16,8 +16,8 @@
 package org.sqlflow.client;
 
 import io.grpc.StatusRuntimeException;
-import org.sqlflow.client.model.RequestHeader;
 import proto.Sqlflow.JobStatus;
+import proto.Sqlflow.Session;
 
 public interface SQLFlow {
   /**
@@ -31,15 +31,21 @@ public interface SQLFlow {
   /**
    * Submit a task to SQLFlow server. This method return immediately.
    *
-   * @param header: specify datasource, user ...
+   * @param session: specify dbConnStr(datasource), user Id ...
+   *     <p>datasource == maxcomputer
+   *     maxcompute://{accesskey_id}:{accesskey_secret}@{endpoint}?curr_project={curr_project}&scheme={scheme}
+   *     <p>datasource == mysql
+   *     mysql://{username}:{password}@tcp({address})/{dbname}[?param1=value1&...&paramN=valueN]
+   *     <p>datasource == hive
+   *     hive://user:password@ip:port/dbname[?auth=<auth_mechanism>&session.<cfg_key1>=<cfg_value1>...&session<cfg_keyN>=valueN]
    * @param sql: sql program.
    *     <p>Example: "SELECT * FROM iris.test; SELECT * FROM iris.iris TO TRAIN DNNClassifier
    *     COLUMN..." *
    * @return return a job id for tracking.
    * @throws IllegalArgumentException header or sql error
    * @throws StatusRuntimeException
    */
-  String submit(RequestHeader header, String sql)
+  String submit(Session session, String sql)
       throws IllegalArgumentException, StatusRuntimeException;
 
   /**
@@ -58,5 +64,5 @@ String submit(RequestHeader header, String sql)
    *
    * @throws InterruptedException thrown by awaitTermination
    */
-  void shutdown() throws InterruptedException;
+  void release() throws InterruptedException;
 }

src/main/java/org/sqlflow/client/impl/SQLFlowImpl.java

@@ -15,13 +15,13 @@
 
 package org.sqlflow.client.impl;
 
+import com.google.common.annotations.VisibleForTesting;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.StatusRuntimeException;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.StringUtils;
 import org.sqlflow.client.SQLFlow;
-import org.sqlflow.client.model.RequestHeader;
 import proto.SQLFlowGrpc;
 import proto.Sqlflow.Job;
 import proto.Sqlflow.JobStatus;
@@ -37,30 +37,21 @@ public void init(String serverUrl) {
     blockingStub = SQLFlowGrpc.newBlockingStub(channel);
   }
 
+  @VisibleForTesting
   public SQLFlowImpl(ManagedChannel channel) {
     this.channel = channel;
     blockingStub = SQLFlowGrpc.newBlockingStub(channel);
   }
 
-  public String submit(RequestHeader header, String sql)
+  public String submit(Session session, String sql)
       throws IllegalArgumentException, StatusRuntimeException {
-    if (header == null || StringUtils.isAnyBlank(header.getDataSource(), header.getUserId())) {
+    if (session == null || StringUtils.isAnyBlank(session.getDbConnStr(), session.getUserId())) {
       throw new IllegalArgumentException("data source and userId are not allowed to be empty");
     }
     if (StringUtils.isBlank(sql)) {
       throw new IllegalArgumentException("sql is empty");
     }
 
-    Session session =
-        Session.newBuilder()
-            .setDbConnStr(header.getDataSource())
-            .setUserId(header.getUserId())
-            .setExitOnSubmit(header.isExitOnSubmit())
-            .setHiveLocation(StringUtils.defaultString(header.getHiveLocation()))
-            .setHdfsNamenodeAddr(StringUtils.defaultString(header.getHdfsNameNode()))
-            .setHdfsUser(StringUtils.defaultString(header.getHdfsUser()))
-            .setHdfsPass(StringUtils.defaultString(header.getHdfsPassword()))
-            .build();
     Request req = Request.newBuilder().setSession(session).setSql(sql).build();
     try {
       Job job = blockingStub.submit(req);
@@ -81,7 +72,7 @@ public JobStatus fetch(String jobId) throws StatusRuntimeException {
     }
   }
 
-  public void shutdown() throws InterruptedException {
+  public void release() throws InterruptedException {
     try {
       channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
     } catch (InterruptedException e) {

src/main/java/org/sqlflow/client/model/RequestHeader.java

@@ -34,7 +34,6 @@
 import org.mockito.ArgumentCaptor;
 import org.mockito.ArgumentMatchers;
 import org.sqlflow.client.impl.SQLFlowImpl;
-import org.sqlflow.client.model.RequestHeader;
 import proto.SQLFlowGrpc;
 import proto.Sqlflow.Job;
 import proto.Sqlflow.JobStatus;
@@ -96,10 +95,12 @@ public void testSubmit() {
     String sql = "SELECT * TO TRAIN DNNClassify WITH ... COLUMN ... INTO ..";
 
     ArgumentCaptor<Request> requestCaptor = ArgumentCaptor.forClass(Request.class);
-    RequestHeader header = new RequestHeader();
-    header.setUserId(userId);
-    header.setDataSource("mysql://root@root@127.0.0.1:3306/iris");
-    String jobId = client.submit(header, sql);
+    Session session =
+        Session.newBuilder()
+            .setUserId(userId)
+            .setDbConnStr("mysql://root@root@127.0.0.1:3306/iris")
+            .build();
+    String jobId = client.submit(session, sql);
     assertEquals(mockJobId(userId, sql), jobId);
     verify(grpcService)
         .submit(requestCaptor.capture(), ArgumentMatchers.<StreamObserver<Job>>any());
@@ -125,6 +126,6 @@ public void testFetch() {
 
   @After
   public void tearDown() throws Exception {
-    client.shutdown();
+    client.release();
   }
 }