hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ser...@apache.org
Subject [34/50] [abbrv] hive git commit: HIVE-14901: HiveServer2: Use user supplied fetch size to determine #rows serialized in tasks (Norris Lee reviewed by Vaibhav Gumashta)
Date Wed, 08 Mar 2017 03:29:12 GMT
HIVE-14901: HiveServer2: Use user supplied fetch size to determine #rows serialized in tasks
(Norris Lee reviewed by Vaibhav Gumashta)


Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/de2a5d63
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/de2a5d63
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/de2a5d63

Branch: refs/heads/hive-14535
Commit: de2a5d63af78ffb6ef97062b06b9f3eb9557e361
Parents: 22af0ef
Author: Vaibhav Gumashta <vgumashta@hortonworks.com>
Authored: Mon Mar 6 14:26:24 2017 -0800
Committer: Vaibhav Gumashta <vgumashta@hortonworks.com>
Committed: Mon Mar 6 14:26:54 2017 -0800

----------------------------------------------------------------------
 .../org/apache/hadoop/hive/conf/HiveConf.java   |  3 +-
 .../apache/hive/jdbc/TestJdbcWithMiniHS2.java   | 30 ++++++++++++++++++++
 .../org/apache/hive/jdbc/HiveConnection.java    | 10 +++++++
 .../serde2/thrift/ThriftJDBCBinarySerDe.java    |  4 ++-
 .../org/apache/hive/service/cli/CLIService.java |  7 ++++-
 .../hive/service/cli/CLIServiceClient.java      |  4 +--
 .../hive/service/cli/session/HiveSession.java   | 10 ++++++-
 .../service/cli/session/HiveSessionImpl.java    | 24 ++++++++++++++++
 .../service/cli/thrift/ThriftCLIService.java    | 19 +++++++++++--
 9 files changed, 103 insertions(+), 8 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/de2a5d63/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
----------------------------------------------------------------------
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index a8aaa5c..676c527 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -2619,7 +2619,7 @@ public class HiveConf extends Configuration {
     // TODO: Make use of this config to configure fetch size
     HIVE_SERVER2_THRIFT_RESULTSET_MAX_FETCH_SIZE("hive.server2.thrift.resultset.max.fetch.size",
         10000, "Max number of rows sent in one Fetch RPC call by the server to the client."),
-    HIVE_SERVER2_RESULTSET_DEFAULT_FETCH_SIZE("hive.server2.resultset.default.fetch.size",
10000,
+    HIVE_SERVER2_THRIFT_RESULTSET_DEFAULT_FETCH_SIZE("hive.server2.thrift.resultset.default.fetch.size",
1000,
         "The number of rows sent in one Fetch RPC call by the server to the client, if not\n"
+
         "specified by the client."),
     HIVE_SERVER2_XSRF_FILTER_ENABLED("hive.server2.xsrf.filter.enabled",false,
@@ -4237,6 +4237,7 @@ public class HiveConf extends Configuration {
     "hive\\.parquet\\..*",
     "hive\\.ppd\\..*",
     "hive\\.prewarm\\..*",
+    "hive\\.server2\\.thrift\\.resultset\\.default\\.fetch\\.size",
     "hive\\.server2\\.proxy\\.user",
     "hive\\.skewjoin\\..*",
     "hive\\.smbjoin\\..*",

http://git-wip-us.apache.org/repos/asf/hive/blob/de2a5d63/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestJdbcWithMiniHS2.java
----------------------------------------------------------------------
diff --git a/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestJdbcWithMiniHS2.java
b/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestJdbcWithMiniHS2.java
index 3d4057b..afe23f8 100644
--- a/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestJdbcWithMiniHS2.java
+++ b/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestJdbcWithMiniHS2.java
@@ -1321,4 +1321,34 @@ public class TestJdbcWithMiniHS2 {
       fs.delete(testPath, true);
     }
   }
+
+  @Test
+  public void testFetchSize() throws Exception {
+    // Test setting fetch size below max
+    Connection fsConn = getConnection(miniHS2.getJdbcURL("default", "fetchSize=50", ""),
+      System.getProperty("user.name"), "bar");
+    Statement stmt = fsConn.createStatement();
+    stmt.execute("set hive.server2.thrift.resultset.serialize.in.tasks=true");
+    int fetchSize = stmt.getFetchSize();
+    assertEquals(50, fetchSize);
+    stmt.close();
+    fsConn.close();
+    // Test setting fetch size above max
+    fsConn = getConnection(
+      miniHS2.getJdbcURL(
+        "default",
+        "fetchSize=" + (miniHS2.getHiveConf().getIntVar(
+          HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_MAX_FETCH_SIZE) + 1),
+        ""),
+      System.getProperty("user.name"), "bar");
+    stmt = fsConn.createStatement();
+    stmt.execute("set hive.server2.thrift.resultset.serialize.in.tasks=true");
+    fetchSize = stmt.getFetchSize();
+    assertEquals(
+      miniHS2.getHiveConf().getIntVar(
+        HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_MAX_FETCH_SIZE),
+      fetchSize);
+    stmt.close();
+    fsConn.close();
+  }
 }

http://git-wip-us.apache.org/repos/asf/hive/blob/de2a5d63/jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java
----------------------------------------------------------------------
diff --git a/jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java b/jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java
index 535ad3d..ed899c6 100644
--- a/jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java
+++ b/jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java
@@ -639,6 +639,9 @@ public class HiveConnection implements java.sql.Connection {
     }
     // switch the database
     openConf.put("use:database", connParams.getDbName());
+    // set the fetchSize
+    openConf.put("set:hiveconf:hive.server2.thrift.resultset.default.fetch.size",
+      Integer.toString(fetchSize));
 
     // set the session configuration
     Map<String, String> sessVars = connParams.getSessionVars();
@@ -664,6 +667,13 @@ public class HiveConnection implements java.sql.Connection {
       }
       protocol = openResp.getServerProtocolVersion();
       sessHandle = openResp.getSessionHandle();
+
+      // Update fetchSize if modified by server
+      String serverFetchSize =
+        openResp.getConfiguration().get("hive.server2.thrift.resultset.default.fetch.size");
+      if (serverFetchSize != null) {
+        fetchSize = Integer.parseInt(serverFetchSize);
+      }
     } catch (TException e) {
       LOG.error("Error opening session", e);
       throw new SQLException("Could not establish connection to "

http://git-wip-us.apache.org/repos/asf/hive/blob/de2a5d63/serde/src/java/org/apache/hadoop/hive/serde2/thrift/ThriftJDBCBinarySerDe.java
----------------------------------------------------------------------
diff --git a/serde/src/java/org/apache/hadoop/hive/serde2/thrift/ThriftJDBCBinarySerDe.java
b/serde/src/java/org/apache/hadoop/hive/serde2/thrift/ThriftJDBCBinarySerDe.java
index 84ed6ba..16cc916 100644
--- a/serde/src/java/org/apache/hadoop/hive/serde2/thrift/ThriftJDBCBinarySerDe.java
+++ b/serde/src/java/org/apache/hadoop/hive/serde2/thrift/ThriftJDBCBinarySerDe.java
@@ -71,7 +71,9 @@ public class ThriftJDBCBinarySerDe extends AbstractSerDe {
   @Override
   public void initialize(Configuration conf, Properties tbl) throws SerDeException {
     // Get column names
-	MAX_BUFFERED_ROWS = HiveConf.getIntVar(conf, HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_MAX_FETCH_SIZE);
+    MAX_BUFFERED_ROWS =
+      HiveConf.getIntVar(conf, HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_DEFAULT_FETCH_SIZE);
+    LOG.info("ThriftJDBCBinarySerDe max number of buffered columns: " + MAX_BUFFERED_ROWS);
     String columnNameProperty = tbl.getProperty(serdeConstants.LIST_COLUMNS);
     String columnTypeProperty = tbl.getProperty(serdeConstants.LIST_COLUMN_TYPES);
     final String columnNameDelimiter = tbl.containsKey(serdeConstants.COLUMN_NAME_DELIMITER)
? tbl

http://git-wip-us.apache.org/repos/asf/hive/blob/de2a5d63/service/src/java/org/apache/hive/service/cli/CLIService.java
----------------------------------------------------------------------
diff --git a/service/src/java/org/apache/hive/service/cli/CLIService.java b/service/src/java/org/apache/hive/service/cli/CLIService.java
index a009e25..5bb4ee0 100644
--- a/service/src/java/org/apache/hive/service/cli/CLIService.java
+++ b/service/src/java/org/apache/hive/service/cli/CLIService.java
@@ -81,7 +81,7 @@ public class CLIService extends CompositeService implements ICLIService
{
   public synchronized void init(HiveConf hiveConf) {
     this.hiveConf = hiveConf;
     sessionManager = new SessionManager(hiveServer2);
-    defaultFetchRows = hiveConf.getIntVar(ConfVars.HIVE_SERVER2_RESULTSET_DEFAULT_FETCH_SIZE);
+    defaultFetchRows = hiveConf.getIntVar(ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_DEFAULT_FETCH_SIZE);
     addService(sessionManager);
     //  If the hadoop cluster is secure, do a kerberos login for the service from the keytab
     if (UserGroupInformation.isSecurityEnabled()) {
@@ -476,6 +476,11 @@ public class CLIService extends CompositeService implements ICLIService
{
     return opStatus;
   }
 
+  public HiveConf getSessionConf(SessionHandle sessionHandle)
+      throws HiveSQLException {
+	  return sessionManager.getSession(sessionHandle).getSessionConf();
+  }
+
   private static final long PROGRESS_MAX_WAIT_NS = 30 * 1000000000l;
   private JobProgressUpdate progressUpdateLog(boolean isProgressLogRequested, Operation operation)
{
     if (!isProgressLogRequested || !ServiceUtils.canProvideProgressLog(hiveConf)

http://git-wip-us.apache.org/repos/asf/hive/blob/de2a5d63/service/src/java/org/apache/hive/service/cli/CLIServiceClient.java
----------------------------------------------------------------------
diff --git a/service/src/java/org/apache/hive/service/cli/CLIServiceClient.java b/service/src/java/org/apache/hive/service/cli/CLIServiceClient.java
index 4b84872..c965abc 100644
--- a/service/src/java/org/apache/hive/service/cli/CLIServiceClient.java
+++ b/service/src/java/org/apache/hive/service/cli/CLIServiceClient.java
@@ -31,10 +31,10 @@ import org.apache.hive.service.auth.HiveAuthFactory;
  *
  */
 public abstract class CLIServiceClient implements ICLIService {
-  protected int defaultFetchRows = ConfVars.HIVE_SERVER2_RESULTSET_DEFAULT_FETCH_SIZE.defaultIntVal;
+  protected int defaultFetchRows = ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_DEFAULT_FETCH_SIZE.defaultIntVal;
 
   public CLIServiceClient(Configuration conf) {
-    defaultFetchRows = HiveConf.getIntVar(conf, ConfVars.HIVE_SERVER2_RESULTSET_DEFAULT_FETCH_SIZE);
+    defaultFetchRows = HiveConf.getIntVar(conf, ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_DEFAULT_FETCH_SIZE);
   }
 
   public SessionHandle openSession(String username, String password)

http://git-wip-us.apache.org/repos/asf/hive/blob/de2a5d63/service/src/java/org/apache/hive/service/cli/session/HiveSession.java
----------------------------------------------------------------------
diff --git a/service/src/java/org/apache/hive/service/cli/session/HiveSession.java b/service/src/java/org/apache/hive/service/cli/session/HiveSession.java
index e5d865b..bd4d90d 100644
--- a/service/src/java/org/apache/hive/service/cli/session/HiveSession.java
+++ b/service/src/java/org/apache/hive/service/cli/session/HiveSession.java
@@ -22,6 +22,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Future;
 
+import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.IMetaStoreClient;
 import org.apache.hadoop.hive.ql.metadata.Hive;
 import org.apache.hive.service.auth.HiveAuthFactory;
@@ -183,7 +184,14 @@ public interface HiveSession extends HiveSessionBase {
 	      String primarySchema, String primaryTable, String foreignCatalog,
 	      String foreignSchema, String foreignTable) 
     throws HiveSQLException;
-  
+
+  /**
+   *
+   * @return
+   * @throws HiveSQLException
+   */
+  HiveConf getSessionConf() throws HiveSQLException;
+
   /**
    * close the session
    * @throws HiveSQLException

http://git-wip-us.apache.org/repos/asf/hive/blob/de2a5d63/service/src/java/org/apache/hive/service/cli/session/HiveSessionImpl.java
----------------------------------------------------------------------
diff --git a/service/src/java/org/apache/hive/service/cli/session/HiveSessionImpl.java b/service/src/java/org/apache/hive/service/cli/session/HiveSessionImpl.java
index f939a93..fd74d55 100644
--- a/service/src/java/org/apache/hive/service/cli/session/HiveSessionImpl.java
+++ b/service/src/java/org/apache/hive/service/cli/session/HiveSessionImpl.java
@@ -177,6 +177,9 @@ public class HiveSessionImpl implements HiveSession {
     }
     // Process global init file: .hiverc
     processGlobalInitFile();
+    // Set fetch size in session conf map
+    sessionConfMap = setFetchSize(sessionConfMap);
+
     if (sessionConfMap != null) {
       configureSession(sessionConfMap);
     }
@@ -245,6 +248,22 @@ public class HiveSessionImpl implements HiveSession {
     }
   }
 
+  private Map<String, String> setFetchSize(Map<String, String> sessionConfMap)
{
+    int maxFetchSize =
+      sessionConf.getIntVar(HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_MAX_FETCH_SIZE);
+    String confFetchSize = sessionConfMap != null ?
+      sessionConfMap.get(
+        "set:hiveconf:" + HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_DEFAULT_FETCH_SIZE.varname)
:
+        null;
+    if (confFetchSize != null && !confFetchSize.isEmpty()) {
+        int fetchSize = Integer.parseInt(confFetchSize);
+        sessionConfMap.put(
+          "set:hiveconf:" + HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_DEFAULT_FETCH_SIZE.varname,
+          Integer.toString(fetchSize > maxFetchSize ? maxFetchSize : fetchSize));
+    }
+    return sessionConfMap;
+  }
+
   private void configureSession(Map<String, String> sessionConfMap) throws HiveSQLException
{
     SessionState.setCurrentSessionState(sessionState);
     for (Map.Entry<String, String> entry : sessionConfMap.entrySet()) {
@@ -465,6 +484,11 @@ public class HiveSessionImpl implements HiveSession {
   }
 
   @Override
+  public HiveConf getSessionConf() throws HiveSQLException {
+	  return this.sessionConf;
+  }
+
+  @Override
   public OperationHandle executeStatement(String statement, Map<String, String> confOverlay)
throws HiveSQLException {
     return executeStatementInternal(statement, confOverlay, false, 0);
   }

http://git-wip-us.apache.org/repos/asf/hive/blob/de2a5d63/service/src/java/org/apache/hive/service/cli/thrift/ThriftCLIService.java
----------------------------------------------------------------------
diff --git a/service/src/java/org/apache/hive/service/cli/thrift/ThriftCLIService.java b/service/src/java/org/apache/hive/service/cli/thrift/ThriftCLIService.java
index 211b33b..0fdc8d9 100644
--- a/service/src/java/org/apache/hive/service/cli/thrift/ThriftCLIService.java
+++ b/service/src/java/org/apache/hive/service/cli/thrift/ThriftCLIService.java
@@ -317,10 +317,19 @@ public abstract class ThriftCLIService extends AbstractService implements
TCLISe
     LOG.info("Client protocol version: " + req.getClient_protocol());
     TOpenSessionResp resp = new TOpenSessionResp();
     try {
+      Map<String, String> openConf = req.getConfiguration();
+
       SessionHandle sessionHandle = getSessionHandle(req, resp);
       resp.setSessionHandle(sessionHandle.toTSessionHandle());
-      // TODO: set real configuration map
-      resp.setConfiguration(new HashMap<String, String>());
+      Map<String, String> configurationMap = new HashMap<String, String>();
+      // Set the updated fetch size from the server into the configuration map for the client
+      HiveConf sessionConf = cliService.getSessionConf(sessionHandle);
+      configurationMap.put(
+        HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_DEFAULT_FETCH_SIZE.varname,
+        Integer.toString(sessionConf != null ?
+          sessionConf.getIntVar(HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_DEFAULT_FETCH_SIZE)
:
+          hiveConf.getIntVar(HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_DEFAULT_FETCH_SIZE)));
+      resp.setConfiguration(configurationMap);
       resp.setStatus(OK_STATUS);
       ThriftCLIServerContext context =
         (ThriftCLIServerContext)currentServerContext.get();
@@ -733,6 +742,12 @@ public abstract class ThriftCLIService extends AbstractService implements
TCLISe
   public TFetchResultsResp FetchResults(TFetchResultsReq req) throws TException {
     TFetchResultsResp resp = new TFetchResultsResp();
     try {
+      // Set fetch size
+      int maxFetchSize =
+        hiveConf.getIntVar(HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_MAX_FETCH_SIZE);
+      if (req.getMaxRows() > maxFetchSize) {
+        req.setMaxRows(maxFetchSize);
+      }
       RowSet rowSet = cliService.fetchResults(
           new OperationHandle(req.getOperationHandle()),
           FetchOrientation.getFetchOrientation(req.getOrientation()),


Mime
View raw message