You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@tajo.apache.org by babokim <gi...@git.apache.org> on 2014/11/24 05:54:30 UTC
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
GitHub user babokim opened a pull request:
https://github.com/apache/tajo/pull/268
TAJO-1206: Implements Thrift proxy server.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/babokim/tajo TAJO-1206
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/tajo/pull/268.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #268
----
commit 80afe993b82d57582fbeab64d20199f4dfa3d9af
Author: jhkim <jh...@apache.org>
Date: 2014-11-21T07:23:32Z
TAJO-1205: Remove possible memory leak in TajoMaster. (jinho)
Closes #265
commit 05ba2b5f82ba8b24e906f662a7cb2e41e463e647
Author: HyoungJun Kim <ba...@babokim-macbook-pro.local>
Date: 2014-11-21T11:47:32Z
TAJO-1026: Implements Thrift proxy server.
commit 68d9dbdfb77e30e6ac65c9984dbc907e989d60a9
Author: HyoungJun Kim <ba...@babokim-macbook-pro.local>
Date: 2014-11-21T16:21:27Z
TAJO-1206: Implements Thrift proxy server.
Add some debug message for "Caused by: java.net.SocketException: Invalid argument" error.
commit bf902fb1e43b9eafe2615fcce915ead5acb72c29
Author: HyoungJun Kim <ba...@babokim-macbook-pro.local>
Date: 2014-11-24T04:35:25Z
TAJO-1206: Implements Thrift proxy server.
commit 0ff4ddffbca31ad34a3203fea57bd913608b492e
Author: HyoungJun Kim <ba...@babokim-macbook-pro.local>
Date: 2014-11-24T04:37:14Z
Merge branch 'master' of https://git-wip-us.apache.org/repos/asf/tajo into TAJO-1206
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by blrunner <gi...@git.apache.org>.
Github user blrunner commented on a diff in the pull request:
https://github.com/apache/tajo/pull/268#discussion_r21014851
--- Diff: tajo-thrift-server/src/main/java/org/apache/tajo/thrift/TajoThriftServiceImpl.java ---
@@ -0,0 +1,1013 @@
+/**
+ * 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.tajo.thrift;
+
+import com.google.protobuf.ByteString;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tajo.*;
+import org.apache.tajo.TajoIdProtos.SessionIdProto;
+import org.apache.tajo.TajoProtos.QueryState;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.client.QueryStatus;
+import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.TajoClientImpl;
+import org.apache.tajo.client.TajoClientUtil;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.ipc.ClientProtos.*;
+import org.apache.tajo.jdbc.FetchResultSet;
+import org.apache.tajo.jdbc.TajoMemoryResultSet;
+import org.apache.tajo.jdbc.TajoResultSet;
+import org.apache.tajo.jdbc.TajoResultSetBase;
+import org.apache.tajo.storage.RowStoreUtil;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.thrift.generated.*;
+import org.apache.tajo.util.TUtil;
+import org.apache.tajo.util.TajoIdUtils;
+import org.apache.thrift.TException;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class TajoThriftServiceImpl implements TajoThriftService.Iface {
+ private static final Log LOG = LogFactory.getLog(TajoThriftServiceImpl.class);
+
+ private Map<TajoIdProtos.SessionIdProto, TajoClientHolder> tajoClientMap =
+ new ConcurrentHashMap<SessionIdProto, TajoClientHolder>();
+
+ private Map<String, ResultSetHolder> queryResultSets = new HashMap<String, ResultSetHolder>();
+ private Map<String, QuerySubmitTask> querySubmitTasks = new HashMap<String, QuerySubmitTask>();
+ private ExecutorService executorService;
+ private ResultSetAndTaskCleaner resultSetAndTaskCleaner;
+
+ private int maxSession;
+ private TajoConf tajoConf;
+
+ public TajoThriftServiceImpl(TajoConf tajoConf) {
+ this.tajoConf = tajoConf;
+ this.maxSession = tajoConf.getInt(ThriftServerConstants.MAX_SESSION_CONF_KEY, 100);
+ int maxTaskRunner = tajoConf.getInt(ThriftServerConstants.MAX_TASK_RUNNER_CONF_KEY, 200);
+
+ this.executorService = Executors.newFixedThreadPool(maxTaskRunner);
+ this.resultSetAndTaskCleaner = new ResultSetAndTaskCleaner();
+
+ this.resultSetAndTaskCleaner.start();
+ }
+
+ public void stop() {
+ if (executorService != null) {
+ executorService.shutdownNow();
+ }
+
+ if (resultSetAndTaskCleaner != null) {
+ resultSetAndTaskCleaner.interrupt();
+ }
+
+ synchronized (tajoClientMap) {
+ for (TajoClientHolder eachClient : tajoClientMap.values()) {
+ eachClient.tajoClient.close();
+ }
+ }
+ }
+
+ public TajoClient getTajoClient(SessionIdProto sessionId) throws TServiceException {
+ if (sessionId == null || !sessionId.hasId()) {
+ throw new TServiceException("No sessionId", "");
+ }
+
+ synchronized (tajoClientMap) {
+ if (tajoClientMap.size() >= maxSession) {
+ throw new TServiceException("exceed max session [" + maxSession + "]", "");
+ }
+ TajoClientHolder tajoClientHolder = tajoClientMap.get(sessionId);
+
+ //if there is multiple proxy server, TajoProxyClient call randomly. So certain proxy server hasn't session.
+ if (tajoClientHolder == null) {
+ //throw new ServiceException("No session info:" + sessionId.getId());
+ try {
+ TajoClient tajoClient = new TajoClientImpl(tajoConf);
+ tajoClient.setSessionId(sessionId);
+ tajoClientHolder = new TajoClientHolder();
+ tajoClientHolder.tajoClient = tajoClient;
+ tajoClientHolder.lastTouchTime = System.currentTimeMillis();
+
+ tajoClientMap.put(sessionId, tajoClientHolder);
+ return tajoClient;
+ } catch (Exception e) {
+ throw new TServiceException(e.getMessage(), StringUtils.stringifyException(e));
+ }
+ } else {
+ tajoClientHolder.lastTouchTime = System.currentTimeMillis();
+ return tajoClientHolder.tajoClient;
+ }
+ }
+ }
+
+ private TServerResponse makeErrorServerResponse(Throwable t) {
+ LOG.error(t.getMessage(), t);
+ TServerResponse response = new TServerResponse();
+ response.setErrorMessage(t.getMessage());
+ response.setResultCode(ResultCode.ERROR.name());
+ response.setBoolResult(false);
+ return response;
+ }
+
+ private TGetQueryStatusResponse makeErrorQueryStatusResponse(String errorMessage) {
+ LOG.error("Query error:" + errorMessage);
+
+ TGetQueryStatusResponse response = new TGetQueryStatusResponse();
+ response.setErrorMessage(errorMessage);
+ response.setState(TajoProtos.QueryState.QUERY_ERROR.name());
+ response.setResultCode(ResultCode.ERROR.name());
+ QueryId queryId = QueryIdFactory.newQueryId(0, 0); //DUMMY
+ response.setQueryId(queryId.toString());
+
+ return response;
+ }
+
+ @Override
+ public TGetQueryStatusResponse submitQuery(String sessionIdStr, String query, boolean isJson) throws TException {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Run Query:" + query);
+ }
+ SessionIdProto sessionId = TajoThriftUtil.makeSessionId(sessionIdStr);
+ TGetQueryStatusResponse queryStatus = new TGetQueryStatusResponse();
+
+ try {
+ TajoClient tajoClient = getTajoClient(sessionId);
+
+ SubmitQueryResponse clientResponse = tajoClient.executeQuery(query);
+
+ if (clientResponse.hasErrorMessage()) {
+ return makeErrorQueryStatusResponse(clientResponse.getErrorMessage());
+ }
+
+ if (clientResponse.getIsForwarded()) {
+ QuerySubmitTask querySubmitTask = new QuerySubmitTask(sessionId);
+ querySubmitTask.queryProgressInfo.queryId = new QueryId(clientResponse.getQueryId());
+
+ QueryStatus clientQueryStatus = tajoClient.getQueryStatus(querySubmitTask.queryProgressInfo.queryId);
+ querySubmitTask.queryProgressInfo.lastTouchTime = System.currentTimeMillis();
+
+ queryStatus.setQueryId(clientQueryStatus.getQueryId().toString());
+ queryStatus.setResultCode(ResultCode.OK.name());
+ queryStatus.setState(clientQueryStatus.getState().name());
+ queryStatus.setProgress(queryStatus.getProgress());
+ queryStatus.setSubmitTime(queryStatus.getSubmitTime());
+ queryStatus.setFinishTime(queryStatus.getFinishTime());
+ queryStatus.setHasResult(clientQueryStatus.hasResult());
+
+ if (queryStatus.getErrorMessage() != null) {
+ queryStatus.setErrorMessage(clientQueryStatus.getErrorMessage());
+ }
+
+ if (queryStatus.getQueryMasterHost() != null) {
+ queryStatus.setQueryMasterHost(clientQueryStatus.getQueryMasterHost());
+ queryStatus.setQueryMasterPort(clientQueryStatus.getQueryMasterPort());
+ }
+
+ querySubmitTask.queryProgressInfo.queryStatus = queryStatus;
+ querySubmitTask.queryProgressInfo.query = query;
+
+ synchronized (querySubmitTasks) {
+ LOG.info(querySubmitTask.getKey() + " query started");
+ querySubmitTasks.put(querySubmitTask.getKey(), querySubmitTask);
+ }
+ executorService.submit(querySubmitTask);
+ return querySubmitTask.queryProgressInfo.queryStatus;
+ } else {
+ QueryId queryId = new QueryId(clientResponse.getQueryId());
+
+ queryStatus.setQueryId(queryId.toString());
+ queryStatus.setResultCode(ResultCode.OK.name());
+ queryStatus.setState(TajoProtos.QueryState.QUERY_SUCCEEDED.name());
+ queryStatus.setProgress(1.0f);
+ queryStatus.setSubmitTime(System.currentTimeMillis());
+ queryStatus.setFinishTime(System.currentTimeMillis());
+ queryStatus.setHasResult(true);
+
+ //select * from table limit 100 or select 1+1
--- End diff --
it looks like a comment for debugging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by babokim <gi...@git.apache.org>.
Github user babokim commented on a diff in the pull request:
https://github.com/apache/tajo/pull/268#discussion_r21415697
--- Diff: tajo-thrift-server/src/main/java/org/apache/tajo/thrift/TajoThriftServiceImpl.java ---
@@ -0,0 +1,1013 @@
+/**
+ * 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.tajo.thrift;
+
+import com.google.protobuf.ByteString;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tajo.*;
+import org.apache.tajo.TajoIdProtos.SessionIdProto;
+import org.apache.tajo.TajoProtos.QueryState;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.client.QueryStatus;
+import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.TajoClientImpl;
+import org.apache.tajo.client.TajoClientUtil;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.ipc.ClientProtos.*;
+import org.apache.tajo.jdbc.FetchResultSet;
+import org.apache.tajo.jdbc.TajoMemoryResultSet;
+import org.apache.tajo.jdbc.TajoResultSet;
+import org.apache.tajo.jdbc.TajoResultSetBase;
+import org.apache.tajo.storage.RowStoreUtil;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.thrift.generated.*;
+import org.apache.tajo.util.TUtil;
+import org.apache.tajo.util.TajoIdUtils;
+import org.apache.thrift.TException;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class TajoThriftServiceImpl implements TajoThriftService.Iface {
+ private static final Log LOG = LogFactory.getLog(TajoThriftServiceImpl.class);
+
+ private Map<TajoIdProtos.SessionIdProto, TajoClientHolder> tajoClientMap =
+ new ConcurrentHashMap<SessionIdProto, TajoClientHolder>();
+
+ private Map<String, ResultSetHolder> queryResultSets = new HashMap<String, ResultSetHolder>();
+ private Map<String, QuerySubmitTask> querySubmitTasks = new HashMap<String, QuerySubmitTask>();
+ private ExecutorService executorService;
+ private ResultSetAndTaskCleaner resultSetAndTaskCleaner;
+
+ private int maxSession;
+ private TajoConf tajoConf;
+
+ public TajoThriftServiceImpl(TajoConf tajoConf) {
+ this.tajoConf = tajoConf;
+ this.maxSession = tajoConf.getInt(ThriftServerConstants.MAX_SESSION_CONF_KEY, 100);
+ int maxTaskRunner = tajoConf.getInt(ThriftServerConstants.MAX_TASK_RUNNER_CONF_KEY, 200);
+
+ this.executorService = Executors.newFixedThreadPool(maxTaskRunner);
+ this.resultSetAndTaskCleaner = new ResultSetAndTaskCleaner();
+
+ this.resultSetAndTaskCleaner.start();
+ }
+
+ public void stop() {
+ if (executorService != null) {
+ executorService.shutdownNow();
+ }
+
+ if (resultSetAndTaskCleaner != null) {
+ resultSetAndTaskCleaner.interrupt();
+ }
+
+ synchronized (tajoClientMap) {
+ for (TajoClientHolder eachClient : tajoClientMap.values()) {
+ eachClient.tajoClient.close();
+ }
+ }
+ }
+
+ public TajoClient getTajoClient(SessionIdProto sessionId) throws TServiceException {
+ if (sessionId == null || !sessionId.hasId()) {
+ throw new TServiceException("No sessionId", "");
+ }
+
+ synchronized (tajoClientMap) {
+ if (tajoClientMap.size() >= maxSession) {
+ throw new TServiceException("exceed max session [" + maxSession + "]", "");
+ }
+ TajoClientHolder tajoClientHolder = tajoClientMap.get(sessionId);
+
+ //if there is multiple proxy server, TajoProxyClient call randomly. So certain proxy server hasn't session.
+ if (tajoClientHolder == null) {
+ //throw new ServiceException("No session info:" + sessionId.getId());
+ try {
+ TajoClient tajoClient = new TajoClientImpl(tajoConf);
+ tajoClient.setSessionId(sessionId);
+ tajoClientHolder = new TajoClientHolder();
+ tajoClientHolder.tajoClient = tajoClient;
+ tajoClientHolder.lastTouchTime = System.currentTimeMillis();
+
+ tajoClientMap.put(sessionId, tajoClientHolder);
+ return tajoClient;
+ } catch (Exception e) {
+ throw new TServiceException(e.getMessage(), StringUtils.stringifyException(e));
+ }
+ } else {
+ tajoClientHolder.lastTouchTime = System.currentTimeMillis();
+ return tajoClientHolder.tajoClient;
+ }
+ }
+ }
+
+ private TServerResponse makeErrorServerResponse(Throwable t) {
+ LOG.error(t.getMessage(), t);
+ TServerResponse response = new TServerResponse();
+ response.setErrorMessage(t.getMessage());
+ response.setResultCode(ResultCode.ERROR.name());
+ response.setBoolResult(false);
+ return response;
+ }
+
+ private TGetQueryStatusResponse makeErrorQueryStatusResponse(String errorMessage) {
+ LOG.error("Query error:" + errorMessage);
+
+ TGetQueryStatusResponse response = new TGetQueryStatusResponse();
+ response.setErrorMessage(errorMessage);
+ response.setState(TajoProtos.QueryState.QUERY_ERROR.name());
+ response.setResultCode(ResultCode.ERROR.name());
+ QueryId queryId = QueryIdFactory.newQueryId(0, 0); //DUMMY
+ response.setQueryId(queryId.toString());
+
+ return response;
+ }
+
+ @Override
+ public TGetQueryStatusResponse submitQuery(String sessionIdStr, String query, boolean isJson) throws TException {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Run Query:" + query);
+ }
+ SessionIdProto sessionId = TajoThriftUtil.makeSessionId(sessionIdStr);
+ TGetQueryStatusResponse queryStatus = new TGetQueryStatusResponse();
+
+ try {
+ TajoClient tajoClient = getTajoClient(sessionId);
+
+ SubmitQueryResponse clientResponse = tajoClient.executeQuery(query);
+
+ if (clientResponse.hasErrorMessage()) {
+ return makeErrorQueryStatusResponse(clientResponse.getErrorMessage());
+ }
+
+ if (clientResponse.getIsForwarded()) {
+ QuerySubmitTask querySubmitTask = new QuerySubmitTask(sessionId);
+ querySubmitTask.queryProgressInfo.queryId = new QueryId(clientResponse.getQueryId());
+
+ QueryStatus clientQueryStatus = tajoClient.getQueryStatus(querySubmitTask.queryProgressInfo.queryId);
+ querySubmitTask.queryProgressInfo.lastTouchTime = System.currentTimeMillis();
+
+ queryStatus.setQueryId(clientQueryStatus.getQueryId().toString());
+ queryStatus.setResultCode(ResultCode.OK.name());
+ queryStatus.setState(clientQueryStatus.getState().name());
+ queryStatus.setProgress(queryStatus.getProgress());
+ queryStatus.setSubmitTime(queryStatus.getSubmitTime());
+ queryStatus.setFinishTime(queryStatus.getFinishTime());
+ queryStatus.setHasResult(clientQueryStatus.hasResult());
+
+ if (queryStatus.getErrorMessage() != null) {
+ queryStatus.setErrorMessage(clientQueryStatus.getErrorMessage());
+ }
+
+ if (queryStatus.getQueryMasterHost() != null) {
+ queryStatus.setQueryMasterHost(clientQueryStatus.getQueryMasterHost());
+ queryStatus.setQueryMasterPort(clientQueryStatus.getQueryMasterPort());
+ }
+
+ querySubmitTask.queryProgressInfo.queryStatus = queryStatus;
+ querySubmitTask.queryProgressInfo.query = query;
+
+ synchronized (querySubmitTasks) {
+ LOG.info(querySubmitTask.getKey() + " query started");
+ querySubmitTasks.put(querySubmitTask.getKey(), querySubmitTask);
+ }
+ executorService.submit(querySubmitTask);
+ return querySubmitTask.queryProgressInfo.queryStatus;
+ } else {
+ QueryId queryId = new QueryId(clientResponse.getQueryId());
+
+ queryStatus.setQueryId(queryId.toString());
+ queryStatus.setResultCode(ResultCode.OK.name());
+ queryStatus.setState(TajoProtos.QueryState.QUERY_SUCCEEDED.name());
+ queryStatus.setProgress(1.0f);
+ queryStatus.setSubmitTime(System.currentTimeMillis());
+ queryStatus.setFinishTime(System.currentTimeMillis());
+ queryStatus.setHasResult(true);
+
+ //select * from table limit 100 or select 1+1
--- End diff --
This comment is not for debugging. It shows a example.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by babokim <gi...@git.apache.org>.
Github user babokim commented on the pull request:
https://github.com/apache/tajo/pull/268#issuecomment-65234931
Hi @hyunsik,
I also thought about that. I think Thrift-based TajoCli is not a critical component in Thrift support functionality and changing to use general Java class is time-consuming work.
What do you think about removing Thrift-based TajoCli from the patch?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by babokim <gi...@git.apache.org>.
Github user babokim commented on a diff in the pull request:
https://github.com/apache/tajo/pull/268#discussion_r21415699
--- Diff: tajo-thrift-server/src/main/java/org/apache/tajo/thrift/client/TajoThriftClient.java ---
@@ -0,0 +1,500 @@
+/**
+ * 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.tajo.thrift.client;
+
+import com.google.protobuf.ServiceException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.tajo.QueryIdFactory;
+import org.apache.tajo.TajoProtos.QueryState;
+import org.apache.tajo.annotation.Nullable;
+import org.apache.tajo.annotation.ThreadSafe;
+import org.apache.tajo.client.*;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.ipc.ClientProtos.*;
+import org.apache.tajo.thrift.TajoThriftUtil;
+import org.apache.tajo.thrift.ThriftServerConstants;
+import org.apache.tajo.thrift.generated.*;
+import org.apache.tajo.thrift.generated.TajoThriftService.Client;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.ResultSet;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@ThreadSafe
+public class TajoThriftClient {
+ private final Log LOG = LogFactory.getLog(TajoThriftClient.class);
+
+ protected final TajoConf tajoConf;
+
+ protected String thriftServer;
+
+ protected Client currentClient;
+
+ private final String baseDatabase;
+
+ private final UserGroupInformation userInfo;
+
+ volatile String sessionId;
+
+ private AtomicBoolean closed = new AtomicBoolean(false);
+
+ // Thrift client is thread unsafe. So every call should be synchronized with callMonitor.
+ private Object callMonitor = new Object();
+
+ public TajoThriftClient(TajoConf tajoConf, String thriftServer) throws IOException {
+ this(tajoConf, thriftServer, null);
+
+ }
+
+ /**
+ * Connect to ThriftServer
+ *
+ * @param tajoConf TajoConf
+ * @param thriftServer ThriftServer
+ * @param baseDatabase The base database name. It is case sensitive. If it is null,
+ * the 'default' database will be used.
+ * @throws java.io.IOException
+ */
+ public TajoThriftClient(TajoConf tajoConf, String thriftServer, @Nullable String baseDatabase) throws IOException {
+ this.tajoConf = tajoConf;
+ this.thriftServer = thriftServer;
+ this.baseDatabase = baseDatabase;
+
+ this.userInfo = UserGroupInformation.getCurrentUser();
+
+ synchronized (callMonitor) {
+ makeConnection();
+ }
+ }
+
+ public TajoConf getConf() {
+ return tajoConf;
+ }
+
+ public UserGroupInformation getUserInfo() {
+ return userInfo;
+ }
+
+ protected void makeConnection() throws IOException {
+ // Should be synchronized with callMonitor
+ if (currentClient == null) {
+ String[] tokens = thriftServer.split(":");
+ TTransport transport = new TSocket(tokens[0], Integer.parseInt(tokens[1]));
+ try {
+ transport.open();
+ } catch (Exception e) {
+ LOG.error("Can not make protocol: " + thriftServer + ", " + e.getMessage(), e);
+ throw new IOException("Can not make protocol", e);
+ }
+ currentClient = new TajoThriftService.Client(new TBinaryProtocol(transport));
+ }
+ }
+
+ public boolean createDatabase(final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.createDatabase(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public boolean existDatabase(final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.existDatabase(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public boolean dropDatabase(final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.dropDatabase(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public List<String> getAllDatabaseNames() throws Exception {
+ return new ReconnectThriftServerCallable<List<String>>(currentClient) {
+ public List<String> syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.getAllDatabases(sessionId);
+ }
+ }.withRetries();
+ }
+
+ public boolean existTable(final String tableName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ try {
+ checkSessionAndGet(client);
+ return client.existTable(sessionId, tableName);
+ } catch (Exception e) {
+ abort();
+ throw e;
+ }
+ }
+ }.withRetries();
+ }
+
+ public boolean dropTable(final String tableName) throws Exception {
+ return dropTable(tableName, false);
+ }
+
+ public boolean dropTable(final String tableName, final boolean purge) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.dropTable(sessionId, tableName, purge);
+ }
+ }.withRetries();
+ }
+
+ public List<String> getTableList(@Nullable final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<List<String>>(currentClient) {
+ public List<String> syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.getTableList(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public TTableDesc getTableDesc(final String tableName) throws Exception {
+ return new ReconnectThriftServerCallable<TTableDesc>(currentClient) {
+ public TTableDesc syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.getTableDesc(sessionId, tableName);
+ }
+ }.withRetries();
+ }
+
+ public void closeQuery(String queryId) {
+ try {
+ checkSessionAndGet(currentClient);
+
+ currentClient.closeQuery(sessionId, queryId);
+ } catch (Exception e) {
+ LOG.warn("Fail to close query (qid=" + queryId + ", msg=" + e.getMessage() + ")", e);
+ }
+ }
+
+ public TGetQueryStatusResponse executeQuery(final String sql) throws Exception {
+ return new ReconnectThriftServerCallable<TGetQueryStatusResponse>(currentClient) {
+ public TGetQueryStatusResponse syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.submitQuery(sessionId, sql, false);
+ }
+ }.withRetries();
+ }
+
+ public boolean updateQuery(final String sql) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.updateQuery(sessionId, sql).isBoolResult();
+ }
+ }.withRetries();
+ }
+
+ public ResultSet executeQueryAndGetResult(final String sql) throws ServiceException, IOException {
+ try {
+ TGetQueryStatusResponse response = new ReconnectThriftServerCallable<TGetQueryStatusResponse>(currentClient) {
+ public TGetQueryStatusResponse syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ TGetQueryStatusResponse response = null;
+ try {
+ response = client.submitQuery(sessionId, sql, false);
+ } catch (TServiceException e) {
+ abort();
+ throw new IOException(e.getMessage(), e);
+ } catch (Throwable t) {
+ throw new IOException(t.getMessage(), t);
+ }
+ if (!ResultCode.OK.name().equals(response.getResultCode()) || response.getErrorMessage() != null) {
+ abort();
+ throw new IOException(response.getErrorMessage());
+ }
+ return response;
+ }
+ }.withRetries();
+
+ if (response != null && response.getQueryId() != null) {
+ return this.getQueryResultAndWait(response.getQueryId(), response);
+ } else {
+ return createNullResultSet(QueryIdFactory.NULL_QUERY_ID.toString());
+ }
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ throw new IOException(e.getMessage(), e);
+ }
+ }
+
+ public ResultSet createNullResultSet(String queryId) throws IOException {
+ TGetQueryStatusResponse emptyQueryStatus = new TGetQueryStatusResponse();
+ emptyQueryStatus.setResultCode(ResultCode.OK.name());
+ emptyQueryStatus.setState(QueryState.QUERY_SUCCEEDED.name());
+ emptyQueryStatus.setQueryId(queryId);
+
+ TQueryResult emptyQueryResult = new TQueryResult();
+
+ emptyQueryResult.setRows(Collections.<ByteBuffer>emptyList());
+ return new TajoThriftResultSet(this, queryId, emptyQueryResult);
+ }
+
+ public ResultSet getQueryResultAndWait(String queryId, TGetQueryStatusResponse queryResponse) throws Exception {
+ if (queryResponse.getQueryResult() != null) {
+ //select 1+1
--- End diff --
This comment is not for debugging. It shows a example.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/tajo/pull/268
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by babokim <gi...@git.apache.org>.
Github user babokim commented on a diff in the pull request:
https://github.com/apache/tajo/pull/268#discussion_r21415701
--- Diff: tajo-thrift-server/src/main/java/org/apache/tajo/thrift/client/TajoThriftResultSet.java ---
@@ -0,0 +1,126 @@
+/**
+ * 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.tajo.thrift.client;
+
+import org.apache.tajo.TajoConstants;
+import org.apache.tajo.jdbc.TajoResultSetBase;
+import org.apache.tajo.storage.RowStoreUtil;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.thrift.TajoThriftUtil;
+import org.apache.tajo.thrift.generated.TQueryResult;
+import org.apache.tajo.thrift.generated.TTableDesc;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+
+public class TajoThriftResultSet extends TajoResultSetBase {
+ private TajoThriftClient tajoThriftClient;
+ private String queryId;
+ private TQueryResult queryResult;
+ private List<ByteBuffer> rowDatas;
+ private Iterator<ByteBuffer> rowIterator;
+ private int fetchSize;
+ private int maxRows;
+ private long totalFetchRows;
+ private RowStoreUtil.RowStoreDecoder rowDecoder;
+ private TTableDesc tableDesc;
+
+ public TajoThriftResultSet(TajoThriftClient tajoThriftClient, String queryId, TQueryResult queryResult) {
+ super();
+
+ this.tajoThriftClient = tajoThriftClient;
+ this.queryId = queryId;
+ this.queryResult = queryResult;
+
+ this.tableDesc = queryResult.getTableDesc();
+ this.schema = TajoThriftUtil.convertSchema(tableDesc.getSchema());
+
+ this.totalRows = tableDesc.getStats() != null ? tableDesc.getStats().getNumRows() : Integer.MAX_VALUE;
+ if (this.totalRows == TajoConstants.UNKNOWN_ROW_NUMBER) {
+ //Case of select * from table
--- End diff --
This comment is not for debugging. It shows a example.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by babokim <gi...@git.apache.org>.
Github user babokim commented on a diff in the pull request:
https://github.com/apache/tajo/pull/268#discussion_r21415698
--- Diff: tajo-thrift-server/src/main/java/org/apache/tajo/thrift/TajoThriftServiceImpl.java ---
@@ -0,0 +1,1013 @@
+/**
+ * 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.tajo.thrift;
+
+import com.google.protobuf.ByteString;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tajo.*;
+import org.apache.tajo.TajoIdProtos.SessionIdProto;
+import org.apache.tajo.TajoProtos.QueryState;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.client.QueryStatus;
+import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.TajoClientImpl;
+import org.apache.tajo.client.TajoClientUtil;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.ipc.ClientProtos.*;
+import org.apache.tajo.jdbc.FetchResultSet;
+import org.apache.tajo.jdbc.TajoMemoryResultSet;
+import org.apache.tajo.jdbc.TajoResultSet;
+import org.apache.tajo.jdbc.TajoResultSetBase;
+import org.apache.tajo.storage.RowStoreUtil;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.thrift.generated.*;
+import org.apache.tajo.util.TUtil;
+import org.apache.tajo.util.TajoIdUtils;
+import org.apache.thrift.TException;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class TajoThriftServiceImpl implements TajoThriftService.Iface {
+ private static final Log LOG = LogFactory.getLog(TajoThriftServiceImpl.class);
+
+ private Map<TajoIdProtos.SessionIdProto, TajoClientHolder> tajoClientMap =
+ new ConcurrentHashMap<SessionIdProto, TajoClientHolder>();
+
+ private Map<String, ResultSetHolder> queryResultSets = new HashMap<String, ResultSetHolder>();
+ private Map<String, QuerySubmitTask> querySubmitTasks = new HashMap<String, QuerySubmitTask>();
+ private ExecutorService executorService;
+ private ResultSetAndTaskCleaner resultSetAndTaskCleaner;
+
+ private int maxSession;
+ private TajoConf tajoConf;
+
+ public TajoThriftServiceImpl(TajoConf tajoConf) {
+ this.tajoConf = tajoConf;
+ this.maxSession = tajoConf.getInt(ThriftServerConstants.MAX_SESSION_CONF_KEY, 100);
+ int maxTaskRunner = tajoConf.getInt(ThriftServerConstants.MAX_TASK_RUNNER_CONF_KEY, 200);
+
+ this.executorService = Executors.newFixedThreadPool(maxTaskRunner);
+ this.resultSetAndTaskCleaner = new ResultSetAndTaskCleaner();
+
+ this.resultSetAndTaskCleaner.start();
+ }
+
+ public void stop() {
+ if (executorService != null) {
+ executorService.shutdownNow();
+ }
+
+ if (resultSetAndTaskCleaner != null) {
+ resultSetAndTaskCleaner.interrupt();
+ }
+
+ synchronized (tajoClientMap) {
+ for (TajoClientHolder eachClient : tajoClientMap.values()) {
+ eachClient.tajoClient.close();
+ }
+ }
+ }
+
+ public TajoClient getTajoClient(SessionIdProto sessionId) throws TServiceException {
+ if (sessionId == null || !sessionId.hasId()) {
+ throw new TServiceException("No sessionId", "");
+ }
+
+ synchronized (tajoClientMap) {
+ if (tajoClientMap.size() >= maxSession) {
+ throw new TServiceException("exceed max session [" + maxSession + "]", "");
+ }
+ TajoClientHolder tajoClientHolder = tajoClientMap.get(sessionId);
+
+ //if there is multiple proxy server, TajoProxyClient call randomly. So certain proxy server hasn't session.
+ if (tajoClientHolder == null) {
+ //throw new ServiceException("No session info:" + sessionId.getId());
+ try {
+ TajoClient tajoClient = new TajoClientImpl(tajoConf);
+ tajoClient.setSessionId(sessionId);
+ tajoClientHolder = new TajoClientHolder();
+ tajoClientHolder.tajoClient = tajoClient;
+ tajoClientHolder.lastTouchTime = System.currentTimeMillis();
+
+ tajoClientMap.put(sessionId, tajoClientHolder);
+ return tajoClient;
+ } catch (Exception e) {
+ throw new TServiceException(e.getMessage(), StringUtils.stringifyException(e));
+ }
+ } else {
+ tajoClientHolder.lastTouchTime = System.currentTimeMillis();
+ return tajoClientHolder.tajoClient;
+ }
+ }
+ }
+
+ private TServerResponse makeErrorServerResponse(Throwable t) {
+ LOG.error(t.getMessage(), t);
+ TServerResponse response = new TServerResponse();
+ response.setErrorMessage(t.getMessage());
+ response.setResultCode(ResultCode.ERROR.name());
+ response.setBoolResult(false);
+ return response;
+ }
+
+ private TGetQueryStatusResponse makeErrorQueryStatusResponse(String errorMessage) {
+ LOG.error("Query error:" + errorMessage);
+
+ TGetQueryStatusResponse response = new TGetQueryStatusResponse();
+ response.setErrorMessage(errorMessage);
+ response.setState(TajoProtos.QueryState.QUERY_ERROR.name());
+ response.setResultCode(ResultCode.ERROR.name());
+ QueryId queryId = QueryIdFactory.newQueryId(0, 0); //DUMMY
+ response.setQueryId(queryId.toString());
+
+ return response;
+ }
+
+ @Override
+ public TGetQueryStatusResponse submitQuery(String sessionIdStr, String query, boolean isJson) throws TException {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Run Query:" + query);
+ }
+ SessionIdProto sessionId = TajoThriftUtil.makeSessionId(sessionIdStr);
+ TGetQueryStatusResponse queryStatus = new TGetQueryStatusResponse();
+
+ try {
+ TajoClient tajoClient = getTajoClient(sessionId);
+
+ SubmitQueryResponse clientResponse = tajoClient.executeQuery(query);
+
+ if (clientResponse.hasErrorMessage()) {
+ return makeErrorQueryStatusResponse(clientResponse.getErrorMessage());
+ }
+
+ if (clientResponse.getIsForwarded()) {
+ QuerySubmitTask querySubmitTask = new QuerySubmitTask(sessionId);
+ querySubmitTask.queryProgressInfo.queryId = new QueryId(clientResponse.getQueryId());
+
+ QueryStatus clientQueryStatus = tajoClient.getQueryStatus(querySubmitTask.queryProgressInfo.queryId);
+ querySubmitTask.queryProgressInfo.lastTouchTime = System.currentTimeMillis();
+
+ queryStatus.setQueryId(clientQueryStatus.getQueryId().toString());
+ queryStatus.setResultCode(ResultCode.OK.name());
+ queryStatus.setState(clientQueryStatus.getState().name());
+ queryStatus.setProgress(queryStatus.getProgress());
+ queryStatus.setSubmitTime(queryStatus.getSubmitTime());
+ queryStatus.setFinishTime(queryStatus.getFinishTime());
+ queryStatus.setHasResult(clientQueryStatus.hasResult());
+
+ if (queryStatus.getErrorMessage() != null) {
+ queryStatus.setErrorMessage(clientQueryStatus.getErrorMessage());
+ }
+
+ if (queryStatus.getQueryMasterHost() != null) {
+ queryStatus.setQueryMasterHost(clientQueryStatus.getQueryMasterHost());
+ queryStatus.setQueryMasterPort(clientQueryStatus.getQueryMasterPort());
+ }
+
+ querySubmitTask.queryProgressInfo.queryStatus = queryStatus;
+ querySubmitTask.queryProgressInfo.query = query;
+
+ synchronized (querySubmitTasks) {
+ LOG.info(querySubmitTask.getKey() + " query started");
+ querySubmitTasks.put(querySubmitTask.getKey(), querySubmitTask);
+ }
+ executorService.submit(querySubmitTask);
+ return querySubmitTask.queryProgressInfo.queryStatus;
+ } else {
+ QueryId queryId = new QueryId(clientResponse.getQueryId());
+
+ queryStatus.setQueryId(queryId.toString());
+ queryStatus.setResultCode(ResultCode.OK.name());
+ queryStatus.setState(TajoProtos.QueryState.QUERY_SUCCEEDED.name());
+ queryStatus.setProgress(1.0f);
+ queryStatus.setSubmitTime(System.currentTimeMillis());
+ queryStatus.setFinishTime(System.currentTimeMillis());
+ queryStatus.setHasResult(true);
+
+ //select * from table limit 100 or select 1+1
+ ResultSet resultSet = TajoClientUtil.createResultSet(tajoConf, tajoClient, clientResponse);
+ if (resultSet instanceof FetchResultSet) {
+ //select * from table limit 100
--- End diff --
This comment is not for debugging. It shows a example.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by blrunner <gi...@git.apache.org>.
Github user blrunner commented on a diff in the pull request:
https://github.com/apache/tajo/pull/268#discussion_r21014875
--- Diff: tajo-thrift-server/src/main/java/org/apache/tajo/thrift/client/TajoThriftResultSet.java ---
@@ -0,0 +1,126 @@
+/**
+ * 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.tajo.thrift.client;
+
+import org.apache.tajo.TajoConstants;
+import org.apache.tajo.jdbc.TajoResultSetBase;
+import org.apache.tajo.storage.RowStoreUtil;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.thrift.TajoThriftUtil;
+import org.apache.tajo.thrift.generated.TQueryResult;
+import org.apache.tajo.thrift.generated.TTableDesc;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+
+public class TajoThriftResultSet extends TajoResultSetBase {
+ private TajoThriftClient tajoThriftClient;
+ private String queryId;
+ private TQueryResult queryResult;
+ private List<ByteBuffer> rowDatas;
+ private Iterator<ByteBuffer> rowIterator;
+ private int fetchSize;
+ private int maxRows;
+ private long totalFetchRows;
+ private RowStoreUtil.RowStoreDecoder rowDecoder;
+ private TTableDesc tableDesc;
+
+ public TajoThriftResultSet(TajoThriftClient tajoThriftClient, String queryId, TQueryResult queryResult) {
+ super();
+
+ this.tajoThriftClient = tajoThriftClient;
+ this.queryId = queryId;
+ this.queryResult = queryResult;
+
+ this.tableDesc = queryResult.getTableDesc();
+ this.schema = TajoThriftUtil.convertSchema(tableDesc.getSchema());
+
+ this.totalRows = tableDesc.getStats() != null ? tableDesc.getStats().getNumRows() : Integer.MAX_VALUE;
+ if (this.totalRows == TajoConstants.UNKNOWN_ROW_NUMBER) {
+ //Case of select * from table
--- End diff --
it looks like a comment for debugging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by hyunsik <gi...@git.apache.org>.
Github user hyunsik commented on the pull request:
https://github.com/apache/tajo/pull/268#issuecomment-65235382
@babokim I agree with your suggestion. Please go ahead :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by blrunner <gi...@git.apache.org>.
Github user blrunner commented on a diff in the pull request:
https://github.com/apache/tajo/pull/268#discussion_r21014853
--- Diff: tajo-thrift-server/src/main/java/org/apache/tajo/thrift/TajoThriftServiceImpl.java ---
@@ -0,0 +1,1013 @@
+/**
+ * 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.tajo.thrift;
+
+import com.google.protobuf.ByteString;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tajo.*;
+import org.apache.tajo.TajoIdProtos.SessionIdProto;
+import org.apache.tajo.TajoProtos.QueryState;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.client.QueryStatus;
+import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.TajoClientImpl;
+import org.apache.tajo.client.TajoClientUtil;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.ipc.ClientProtos.*;
+import org.apache.tajo.jdbc.FetchResultSet;
+import org.apache.tajo.jdbc.TajoMemoryResultSet;
+import org.apache.tajo.jdbc.TajoResultSet;
+import org.apache.tajo.jdbc.TajoResultSetBase;
+import org.apache.tajo.storage.RowStoreUtil;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.thrift.generated.*;
+import org.apache.tajo.util.TUtil;
+import org.apache.tajo.util.TajoIdUtils;
+import org.apache.thrift.TException;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class TajoThriftServiceImpl implements TajoThriftService.Iface {
+ private static final Log LOG = LogFactory.getLog(TajoThriftServiceImpl.class);
+
+ private Map<TajoIdProtos.SessionIdProto, TajoClientHolder> tajoClientMap =
+ new ConcurrentHashMap<SessionIdProto, TajoClientHolder>();
+
+ private Map<String, ResultSetHolder> queryResultSets = new HashMap<String, ResultSetHolder>();
+ private Map<String, QuerySubmitTask> querySubmitTasks = new HashMap<String, QuerySubmitTask>();
+ private ExecutorService executorService;
+ private ResultSetAndTaskCleaner resultSetAndTaskCleaner;
+
+ private int maxSession;
+ private TajoConf tajoConf;
+
+ public TajoThriftServiceImpl(TajoConf tajoConf) {
+ this.tajoConf = tajoConf;
+ this.maxSession = tajoConf.getInt(ThriftServerConstants.MAX_SESSION_CONF_KEY, 100);
+ int maxTaskRunner = tajoConf.getInt(ThriftServerConstants.MAX_TASK_RUNNER_CONF_KEY, 200);
+
+ this.executorService = Executors.newFixedThreadPool(maxTaskRunner);
+ this.resultSetAndTaskCleaner = new ResultSetAndTaskCleaner();
+
+ this.resultSetAndTaskCleaner.start();
+ }
+
+ public void stop() {
+ if (executorService != null) {
+ executorService.shutdownNow();
+ }
+
+ if (resultSetAndTaskCleaner != null) {
+ resultSetAndTaskCleaner.interrupt();
+ }
+
+ synchronized (tajoClientMap) {
+ for (TajoClientHolder eachClient : tajoClientMap.values()) {
+ eachClient.tajoClient.close();
+ }
+ }
+ }
+
+ public TajoClient getTajoClient(SessionIdProto sessionId) throws TServiceException {
+ if (sessionId == null || !sessionId.hasId()) {
+ throw new TServiceException("No sessionId", "");
+ }
+
+ synchronized (tajoClientMap) {
+ if (tajoClientMap.size() >= maxSession) {
+ throw new TServiceException("exceed max session [" + maxSession + "]", "");
+ }
+ TajoClientHolder tajoClientHolder = tajoClientMap.get(sessionId);
+
+ //if there is multiple proxy server, TajoProxyClient call randomly. So certain proxy server hasn't session.
+ if (tajoClientHolder == null) {
+ //throw new ServiceException("No session info:" + sessionId.getId());
+ try {
+ TajoClient tajoClient = new TajoClientImpl(tajoConf);
+ tajoClient.setSessionId(sessionId);
+ tajoClientHolder = new TajoClientHolder();
+ tajoClientHolder.tajoClient = tajoClient;
+ tajoClientHolder.lastTouchTime = System.currentTimeMillis();
+
+ tajoClientMap.put(sessionId, tajoClientHolder);
+ return tajoClient;
+ } catch (Exception e) {
+ throw new TServiceException(e.getMessage(), StringUtils.stringifyException(e));
+ }
+ } else {
+ tajoClientHolder.lastTouchTime = System.currentTimeMillis();
+ return tajoClientHolder.tajoClient;
+ }
+ }
+ }
+
+ private TServerResponse makeErrorServerResponse(Throwable t) {
+ LOG.error(t.getMessage(), t);
+ TServerResponse response = new TServerResponse();
+ response.setErrorMessage(t.getMessage());
+ response.setResultCode(ResultCode.ERROR.name());
+ response.setBoolResult(false);
+ return response;
+ }
+
+ private TGetQueryStatusResponse makeErrorQueryStatusResponse(String errorMessage) {
+ LOG.error("Query error:" + errorMessage);
+
+ TGetQueryStatusResponse response = new TGetQueryStatusResponse();
+ response.setErrorMessage(errorMessage);
+ response.setState(TajoProtos.QueryState.QUERY_ERROR.name());
+ response.setResultCode(ResultCode.ERROR.name());
+ QueryId queryId = QueryIdFactory.newQueryId(0, 0); //DUMMY
+ response.setQueryId(queryId.toString());
+
+ return response;
+ }
+
+ @Override
+ public TGetQueryStatusResponse submitQuery(String sessionIdStr, String query, boolean isJson) throws TException {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Run Query:" + query);
+ }
+ SessionIdProto sessionId = TajoThriftUtil.makeSessionId(sessionIdStr);
+ TGetQueryStatusResponse queryStatus = new TGetQueryStatusResponse();
+
+ try {
+ TajoClient tajoClient = getTajoClient(sessionId);
+
+ SubmitQueryResponse clientResponse = tajoClient.executeQuery(query);
+
+ if (clientResponse.hasErrorMessage()) {
+ return makeErrorQueryStatusResponse(clientResponse.getErrorMessage());
+ }
+
+ if (clientResponse.getIsForwarded()) {
+ QuerySubmitTask querySubmitTask = new QuerySubmitTask(sessionId);
+ querySubmitTask.queryProgressInfo.queryId = new QueryId(clientResponse.getQueryId());
+
+ QueryStatus clientQueryStatus = tajoClient.getQueryStatus(querySubmitTask.queryProgressInfo.queryId);
+ querySubmitTask.queryProgressInfo.lastTouchTime = System.currentTimeMillis();
+
+ queryStatus.setQueryId(clientQueryStatus.getQueryId().toString());
+ queryStatus.setResultCode(ResultCode.OK.name());
+ queryStatus.setState(clientQueryStatus.getState().name());
+ queryStatus.setProgress(queryStatus.getProgress());
+ queryStatus.setSubmitTime(queryStatus.getSubmitTime());
+ queryStatus.setFinishTime(queryStatus.getFinishTime());
+ queryStatus.setHasResult(clientQueryStatus.hasResult());
+
+ if (queryStatus.getErrorMessage() != null) {
+ queryStatus.setErrorMessage(clientQueryStatus.getErrorMessage());
+ }
+
+ if (queryStatus.getQueryMasterHost() != null) {
+ queryStatus.setQueryMasterHost(clientQueryStatus.getQueryMasterHost());
+ queryStatus.setQueryMasterPort(clientQueryStatus.getQueryMasterPort());
+ }
+
+ querySubmitTask.queryProgressInfo.queryStatus = queryStatus;
+ querySubmitTask.queryProgressInfo.query = query;
+
+ synchronized (querySubmitTasks) {
+ LOG.info(querySubmitTask.getKey() + " query started");
+ querySubmitTasks.put(querySubmitTask.getKey(), querySubmitTask);
+ }
+ executorService.submit(querySubmitTask);
+ return querySubmitTask.queryProgressInfo.queryStatus;
+ } else {
+ QueryId queryId = new QueryId(clientResponse.getQueryId());
+
+ queryStatus.setQueryId(queryId.toString());
+ queryStatus.setResultCode(ResultCode.OK.name());
+ queryStatus.setState(TajoProtos.QueryState.QUERY_SUCCEEDED.name());
+ queryStatus.setProgress(1.0f);
+ queryStatus.setSubmitTime(System.currentTimeMillis());
+ queryStatus.setFinishTime(System.currentTimeMillis());
+ queryStatus.setHasResult(true);
+
+ //select * from table limit 100 or select 1+1
+ ResultSet resultSet = TajoClientUtil.createResultSet(tajoConf, tajoClient, clientResponse);
+ if (resultSet instanceof FetchResultSet) {
+ //select * from table limit 100
--- End diff --
it looks like a comment for debugging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by blrunner <gi...@git.apache.org>.
Github user blrunner commented on a diff in the pull request:
https://github.com/apache/tajo/pull/268#discussion_r21014869
--- Diff: tajo-thrift-server/src/main/java/org/apache/tajo/thrift/client/TajoThriftClient.java ---
@@ -0,0 +1,500 @@
+/**
+ * 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.tajo.thrift.client;
+
+import com.google.protobuf.ServiceException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.tajo.QueryIdFactory;
+import org.apache.tajo.TajoProtos.QueryState;
+import org.apache.tajo.annotation.Nullable;
+import org.apache.tajo.annotation.ThreadSafe;
+import org.apache.tajo.client.*;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.ipc.ClientProtos.*;
+import org.apache.tajo.thrift.TajoThriftUtil;
+import org.apache.tajo.thrift.ThriftServerConstants;
+import org.apache.tajo.thrift.generated.*;
+import org.apache.tajo.thrift.generated.TajoThriftService.Client;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.ResultSet;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@ThreadSafe
+public class TajoThriftClient {
+ private final Log LOG = LogFactory.getLog(TajoThriftClient.class);
+
+ protected final TajoConf tajoConf;
+
+ protected String thriftServer;
+
+ protected Client currentClient;
+
+ private final String baseDatabase;
+
+ private final UserGroupInformation userInfo;
+
+ volatile String sessionId;
+
+ private AtomicBoolean closed = new AtomicBoolean(false);
+
+ // Thrift client is thread unsafe. So every call should be synchronized with callMonitor.
+ private Object callMonitor = new Object();
+
+ public TajoThriftClient(TajoConf tajoConf, String thriftServer) throws IOException {
+ this(tajoConf, thriftServer, null);
+
+ }
+
+ /**
+ * Connect to ThriftServer
+ *
+ * @param tajoConf TajoConf
+ * @param thriftServer ThriftServer
+ * @param baseDatabase The base database name. It is case sensitive. If it is null,
+ * the 'default' database will be used.
+ * @throws java.io.IOException
+ */
+ public TajoThriftClient(TajoConf tajoConf, String thriftServer, @Nullable String baseDatabase) throws IOException {
+ this.tajoConf = tajoConf;
+ this.thriftServer = thriftServer;
+ this.baseDatabase = baseDatabase;
+
+ this.userInfo = UserGroupInformation.getCurrentUser();
+
+ synchronized (callMonitor) {
+ makeConnection();
+ }
+ }
+
+ public TajoConf getConf() {
+ return tajoConf;
+ }
+
+ public UserGroupInformation getUserInfo() {
+ return userInfo;
+ }
+
+ protected void makeConnection() throws IOException {
+ // Should be synchronized with callMonitor
+ if (currentClient == null) {
+ String[] tokens = thriftServer.split(":");
+ TTransport transport = new TSocket(tokens[0], Integer.parseInt(tokens[1]));
+ try {
+ transport.open();
+ } catch (Exception e) {
+ LOG.error("Can not make protocol: " + thriftServer + ", " + e.getMessage(), e);
+ throw new IOException("Can not make protocol", e);
+ }
+ currentClient = new TajoThriftService.Client(new TBinaryProtocol(transport));
+ }
+ }
+
+ public boolean createDatabase(final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.createDatabase(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public boolean existDatabase(final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.existDatabase(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public boolean dropDatabase(final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.dropDatabase(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public List<String> getAllDatabaseNames() throws Exception {
+ return new ReconnectThriftServerCallable<List<String>>(currentClient) {
+ public List<String> syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.getAllDatabases(sessionId);
+ }
+ }.withRetries();
+ }
+
+ public boolean existTable(final String tableName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ try {
+ checkSessionAndGet(client);
+ return client.existTable(sessionId, tableName);
+ } catch (Exception e) {
+ abort();
+ throw e;
+ }
+ }
+ }.withRetries();
+ }
+
+ public boolean dropTable(final String tableName) throws Exception {
+ return dropTable(tableName, false);
+ }
+
+ public boolean dropTable(final String tableName, final boolean purge) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.dropTable(sessionId, tableName, purge);
+ }
+ }.withRetries();
+ }
+
+ public List<String> getTableList(@Nullable final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<List<String>>(currentClient) {
+ public List<String> syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.getTableList(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public TTableDesc getTableDesc(final String tableName) throws Exception {
+ return new ReconnectThriftServerCallable<TTableDesc>(currentClient) {
+ public TTableDesc syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.getTableDesc(sessionId, tableName);
+ }
+ }.withRetries();
+ }
+
+ public void closeQuery(String queryId) {
+ try {
+ checkSessionAndGet(currentClient);
+
+ currentClient.closeQuery(sessionId, queryId);
+ } catch (Exception e) {
+ LOG.warn("Fail to close query (qid=" + queryId + ", msg=" + e.getMessage() + ")", e);
+ }
+ }
+
+ public TGetQueryStatusResponse executeQuery(final String sql) throws Exception {
+ return new ReconnectThriftServerCallable<TGetQueryStatusResponse>(currentClient) {
+ public TGetQueryStatusResponse syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.submitQuery(sessionId, sql, false);
+ }
+ }.withRetries();
+ }
+
+ public boolean updateQuery(final String sql) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.updateQuery(sessionId, sql).isBoolResult();
+ }
+ }.withRetries();
+ }
+
+ public ResultSet executeQueryAndGetResult(final String sql) throws ServiceException, IOException {
+ try {
+ TGetQueryStatusResponse response = new ReconnectThriftServerCallable<TGetQueryStatusResponse>(currentClient) {
+ public TGetQueryStatusResponse syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ TGetQueryStatusResponse response = null;
+ try {
+ response = client.submitQuery(sessionId, sql, false);
+ } catch (TServiceException e) {
+ abort();
+ throw new IOException(e.getMessage(), e);
+ } catch (Throwable t) {
+ throw new IOException(t.getMessage(), t);
+ }
+ if (!ResultCode.OK.name().equals(response.getResultCode()) || response.getErrorMessage() != null) {
+ abort();
+ throw new IOException(response.getErrorMessage());
+ }
+ return response;
+ }
+ }.withRetries();
+
+ if (response != null && response.getQueryId() != null) {
+ return this.getQueryResultAndWait(response.getQueryId(), response);
+ } else {
+ return createNullResultSet(QueryIdFactory.NULL_QUERY_ID.toString());
+ }
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ throw new IOException(e.getMessage(), e);
+ }
+ }
+
+ public ResultSet createNullResultSet(String queryId) throws IOException {
+ TGetQueryStatusResponse emptyQueryStatus = new TGetQueryStatusResponse();
+ emptyQueryStatus.setResultCode(ResultCode.OK.name());
+ emptyQueryStatus.setState(QueryState.QUERY_SUCCEEDED.name());
+ emptyQueryStatus.setQueryId(queryId);
+
+ TQueryResult emptyQueryResult = new TQueryResult();
+
+ emptyQueryResult.setRows(Collections.<ByteBuffer>emptyList());
+ return new TajoThriftResultSet(this, queryId, emptyQueryResult);
+ }
+
+ public ResultSet getQueryResultAndWait(String queryId, TGetQueryStatusResponse queryResponse) throws Exception {
+ if (queryResponse.getQueryResult() != null) {
+ //select 1+1
--- End diff --
it looks like a comment for debugging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by blrunner <gi...@git.apache.org>.
Github user blrunner commented on a diff in the pull request:
https://github.com/apache/tajo/pull/268#discussion_r21014872
--- Diff: tajo-thrift-server/src/main/java/org/apache/tajo/thrift/client/TajoThriftClient.java ---
@@ -0,0 +1,500 @@
+/**
+ * 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.tajo.thrift.client;
+
+import com.google.protobuf.ServiceException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.tajo.QueryIdFactory;
+import org.apache.tajo.TajoProtos.QueryState;
+import org.apache.tajo.annotation.Nullable;
+import org.apache.tajo.annotation.ThreadSafe;
+import org.apache.tajo.client.*;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.ipc.ClientProtos.*;
+import org.apache.tajo.thrift.TajoThriftUtil;
+import org.apache.tajo.thrift.ThriftServerConstants;
+import org.apache.tajo.thrift.generated.*;
+import org.apache.tajo.thrift.generated.TajoThriftService.Client;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.ResultSet;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@ThreadSafe
+public class TajoThriftClient {
+ private final Log LOG = LogFactory.getLog(TajoThriftClient.class);
+
+ protected final TajoConf tajoConf;
+
+ protected String thriftServer;
+
+ protected Client currentClient;
+
+ private final String baseDatabase;
+
+ private final UserGroupInformation userInfo;
+
+ volatile String sessionId;
+
+ private AtomicBoolean closed = new AtomicBoolean(false);
+
+ // Thrift client is thread unsafe. So every call should be synchronized with callMonitor.
+ private Object callMonitor = new Object();
+
+ public TajoThriftClient(TajoConf tajoConf, String thriftServer) throws IOException {
+ this(tajoConf, thriftServer, null);
+
+ }
+
+ /**
+ * Connect to ThriftServer
+ *
+ * @param tajoConf TajoConf
+ * @param thriftServer ThriftServer
+ * @param baseDatabase The base database name. It is case sensitive. If it is null,
+ * the 'default' database will be used.
+ * @throws java.io.IOException
+ */
+ public TajoThriftClient(TajoConf tajoConf, String thriftServer, @Nullable String baseDatabase) throws IOException {
+ this.tajoConf = tajoConf;
+ this.thriftServer = thriftServer;
+ this.baseDatabase = baseDatabase;
+
+ this.userInfo = UserGroupInformation.getCurrentUser();
+
+ synchronized (callMonitor) {
+ makeConnection();
+ }
+ }
+
+ public TajoConf getConf() {
+ return tajoConf;
+ }
+
+ public UserGroupInformation getUserInfo() {
+ return userInfo;
+ }
+
+ protected void makeConnection() throws IOException {
+ // Should be synchronized with callMonitor
+ if (currentClient == null) {
+ String[] tokens = thriftServer.split(":");
+ TTransport transport = new TSocket(tokens[0], Integer.parseInt(tokens[1]));
+ try {
+ transport.open();
+ } catch (Exception e) {
+ LOG.error("Can not make protocol: " + thriftServer + ", " + e.getMessage(), e);
+ throw new IOException("Can not make protocol", e);
+ }
+ currentClient = new TajoThriftService.Client(new TBinaryProtocol(transport));
+ }
+ }
+
+ public boolean createDatabase(final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.createDatabase(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public boolean existDatabase(final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.existDatabase(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public boolean dropDatabase(final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.dropDatabase(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public List<String> getAllDatabaseNames() throws Exception {
+ return new ReconnectThriftServerCallable<List<String>>(currentClient) {
+ public List<String> syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.getAllDatabases(sessionId);
+ }
+ }.withRetries();
+ }
+
+ public boolean existTable(final String tableName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ try {
+ checkSessionAndGet(client);
+ return client.existTable(sessionId, tableName);
+ } catch (Exception e) {
+ abort();
+ throw e;
+ }
+ }
+ }.withRetries();
+ }
+
+ public boolean dropTable(final String tableName) throws Exception {
+ return dropTable(tableName, false);
+ }
+
+ public boolean dropTable(final String tableName, final boolean purge) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.dropTable(sessionId, tableName, purge);
+ }
+ }.withRetries();
+ }
+
+ public List<String> getTableList(@Nullable final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<List<String>>(currentClient) {
+ public List<String> syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.getTableList(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public TTableDesc getTableDesc(final String tableName) throws Exception {
+ return new ReconnectThriftServerCallable<TTableDesc>(currentClient) {
+ public TTableDesc syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.getTableDesc(sessionId, tableName);
+ }
+ }.withRetries();
+ }
+
+ public void closeQuery(String queryId) {
+ try {
+ checkSessionAndGet(currentClient);
+
+ currentClient.closeQuery(sessionId, queryId);
+ } catch (Exception e) {
+ LOG.warn("Fail to close query (qid=" + queryId + ", msg=" + e.getMessage() + ")", e);
+ }
+ }
+
+ public TGetQueryStatusResponse executeQuery(final String sql) throws Exception {
+ return new ReconnectThriftServerCallable<TGetQueryStatusResponse>(currentClient) {
+ public TGetQueryStatusResponse syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.submitQuery(sessionId, sql, false);
+ }
+ }.withRetries();
+ }
+
+ public boolean updateQuery(final String sql) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.updateQuery(sessionId, sql).isBoolResult();
+ }
+ }.withRetries();
+ }
+
+ public ResultSet executeQueryAndGetResult(final String sql) throws ServiceException, IOException {
+ try {
+ TGetQueryStatusResponse response = new ReconnectThriftServerCallable<TGetQueryStatusResponse>(currentClient) {
+ public TGetQueryStatusResponse syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ TGetQueryStatusResponse response = null;
+ try {
+ response = client.submitQuery(sessionId, sql, false);
+ } catch (TServiceException e) {
+ abort();
+ throw new IOException(e.getMessage(), e);
+ } catch (Throwable t) {
+ throw new IOException(t.getMessage(), t);
+ }
+ if (!ResultCode.OK.name().equals(response.getResultCode()) || response.getErrorMessage() != null) {
+ abort();
+ throw new IOException(response.getErrorMessage());
+ }
+ return response;
+ }
+ }.withRetries();
+
+ if (response != null && response.getQueryId() != null) {
+ return this.getQueryResultAndWait(response.getQueryId(), response);
+ } else {
+ return createNullResultSet(QueryIdFactory.NULL_QUERY_ID.toString());
+ }
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ throw new IOException(e.getMessage(), e);
+ }
+ }
+
+ public ResultSet createNullResultSet(String queryId) throws IOException {
+ TGetQueryStatusResponse emptyQueryStatus = new TGetQueryStatusResponse();
+ emptyQueryStatus.setResultCode(ResultCode.OK.name());
+ emptyQueryStatus.setState(QueryState.QUERY_SUCCEEDED.name());
+ emptyQueryStatus.setQueryId(queryId);
+
+ TQueryResult emptyQueryResult = new TQueryResult();
+
+ emptyQueryResult.setRows(Collections.<ByteBuffer>emptyList());
+ return new TajoThriftResultSet(this, queryId, emptyQueryResult);
+ }
+
+ public ResultSet getQueryResultAndWait(String queryId, TGetQueryStatusResponse queryResponse) throws Exception {
+ if (queryResponse.getQueryResult() != null) {
+ //select 1+1
+ TQueryResult queryResult = queryResponse.getQueryResult();
+ ResultSet resultSet = new TajoThriftMemoryResultSet(this, queryId,
+ TajoThriftUtil.convertSchema(queryResult.getSchema()),
+ queryResult.getRows(), queryResult.getRows() == null ? 0 : queryResult.getRows().size());
+ return resultSet;
+ }
+
+ TGetQueryStatusResponse status = getQueryStatus(queryId);
+ while(status != null && TajoThriftUtil.isQueryRunnning(status.getState())) {
+ status = getQueryStatus(queryId);
+ try {
+ //TODO use thread
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ if (QueryState.QUERY_SUCCEEDED.name().equals(status.getState())) {
+ if (status.isHasResult()) {
+ //select * from lineitem
--- End diff --
it looks like a comment for debugging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by hyunsik <gi...@git.apache.org>.
Github user hyunsik commented on the pull request:
https://github.com/apache/tajo/pull/268#issuecomment-65009386
Hi @babokim,
It's really great work. BTW, in my opinion, there are too many duplicated classes in Thrift-based TajoCli. I think that it would be better if we reuse TajoClient code. It could be easily possible if we change some protobuf classes used by TajoClient uses to general Java classes. For it, we may need two converter classes which is from protobuf to java classes and another is from thrift to java classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] tajo pull request: TAJO-1206: Implements Thrift proxy server.
Posted by babokim <gi...@git.apache.org>.
Github user babokim commented on a diff in the pull request:
https://github.com/apache/tajo/pull/268#discussion_r21415700
--- Diff: tajo-thrift-server/src/main/java/org/apache/tajo/thrift/client/TajoThriftClient.java ---
@@ -0,0 +1,500 @@
+/**
+ * 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.tajo.thrift.client;
+
+import com.google.protobuf.ServiceException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.tajo.QueryIdFactory;
+import org.apache.tajo.TajoProtos.QueryState;
+import org.apache.tajo.annotation.Nullable;
+import org.apache.tajo.annotation.ThreadSafe;
+import org.apache.tajo.client.*;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.ipc.ClientProtos.*;
+import org.apache.tajo.thrift.TajoThriftUtil;
+import org.apache.tajo.thrift.ThriftServerConstants;
+import org.apache.tajo.thrift.generated.*;
+import org.apache.tajo.thrift.generated.TajoThriftService.Client;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.ResultSet;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@ThreadSafe
+public class TajoThriftClient {
+ private final Log LOG = LogFactory.getLog(TajoThriftClient.class);
+
+ protected final TajoConf tajoConf;
+
+ protected String thriftServer;
+
+ protected Client currentClient;
+
+ private final String baseDatabase;
+
+ private final UserGroupInformation userInfo;
+
+ volatile String sessionId;
+
+ private AtomicBoolean closed = new AtomicBoolean(false);
+
+ // Thrift client is thread unsafe. So every call should be synchronized with callMonitor.
+ private Object callMonitor = new Object();
+
+ public TajoThriftClient(TajoConf tajoConf, String thriftServer) throws IOException {
+ this(tajoConf, thriftServer, null);
+
+ }
+
+ /**
+ * Connect to ThriftServer
+ *
+ * @param tajoConf TajoConf
+ * @param thriftServer ThriftServer
+ * @param baseDatabase The base database name. It is case sensitive. If it is null,
+ * the 'default' database will be used.
+ * @throws java.io.IOException
+ */
+ public TajoThriftClient(TajoConf tajoConf, String thriftServer, @Nullable String baseDatabase) throws IOException {
+ this.tajoConf = tajoConf;
+ this.thriftServer = thriftServer;
+ this.baseDatabase = baseDatabase;
+
+ this.userInfo = UserGroupInformation.getCurrentUser();
+
+ synchronized (callMonitor) {
+ makeConnection();
+ }
+ }
+
+ public TajoConf getConf() {
+ return tajoConf;
+ }
+
+ public UserGroupInformation getUserInfo() {
+ return userInfo;
+ }
+
+ protected void makeConnection() throws IOException {
+ // Should be synchronized with callMonitor
+ if (currentClient == null) {
+ String[] tokens = thriftServer.split(":");
+ TTransport transport = new TSocket(tokens[0], Integer.parseInt(tokens[1]));
+ try {
+ transport.open();
+ } catch (Exception e) {
+ LOG.error("Can not make protocol: " + thriftServer + ", " + e.getMessage(), e);
+ throw new IOException("Can not make protocol", e);
+ }
+ currentClient = new TajoThriftService.Client(new TBinaryProtocol(transport));
+ }
+ }
+
+ public boolean createDatabase(final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.createDatabase(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public boolean existDatabase(final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.existDatabase(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public boolean dropDatabase(final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.dropDatabase(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public List<String> getAllDatabaseNames() throws Exception {
+ return new ReconnectThriftServerCallable<List<String>>(currentClient) {
+ public List<String> syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.getAllDatabases(sessionId);
+ }
+ }.withRetries();
+ }
+
+ public boolean existTable(final String tableName) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ try {
+ checkSessionAndGet(client);
+ return client.existTable(sessionId, tableName);
+ } catch (Exception e) {
+ abort();
+ throw e;
+ }
+ }
+ }.withRetries();
+ }
+
+ public boolean dropTable(final String tableName) throws Exception {
+ return dropTable(tableName, false);
+ }
+
+ public boolean dropTable(final String tableName, final boolean purge) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.dropTable(sessionId, tableName, purge);
+ }
+ }.withRetries();
+ }
+
+ public List<String> getTableList(@Nullable final String databaseName) throws Exception {
+ return new ReconnectThriftServerCallable<List<String>>(currentClient) {
+ public List<String> syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.getTableList(sessionId, databaseName);
+ }
+ }.withRetries();
+ }
+
+ public TTableDesc getTableDesc(final String tableName) throws Exception {
+ return new ReconnectThriftServerCallable<TTableDesc>(currentClient) {
+ public TTableDesc syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.getTableDesc(sessionId, tableName);
+ }
+ }.withRetries();
+ }
+
+ public void closeQuery(String queryId) {
+ try {
+ checkSessionAndGet(currentClient);
+
+ currentClient.closeQuery(sessionId, queryId);
+ } catch (Exception e) {
+ LOG.warn("Fail to close query (qid=" + queryId + ", msg=" + e.getMessage() + ")", e);
+ }
+ }
+
+ public TGetQueryStatusResponse executeQuery(final String sql) throws Exception {
+ return new ReconnectThriftServerCallable<TGetQueryStatusResponse>(currentClient) {
+ public TGetQueryStatusResponse syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.submitQuery(sessionId, sql, false);
+ }
+ }.withRetries();
+ }
+
+ public boolean updateQuery(final String sql) throws Exception {
+ return new ReconnectThriftServerCallable<Boolean>(currentClient) {
+ public Boolean syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ return client.updateQuery(sessionId, sql).isBoolResult();
+ }
+ }.withRetries();
+ }
+
+ public ResultSet executeQueryAndGetResult(final String sql) throws ServiceException, IOException {
+ try {
+ TGetQueryStatusResponse response = new ReconnectThriftServerCallable<TGetQueryStatusResponse>(currentClient) {
+ public TGetQueryStatusResponse syncCall(Client client) throws Exception {
+ checkSessionAndGet(client);
+ TGetQueryStatusResponse response = null;
+ try {
+ response = client.submitQuery(sessionId, sql, false);
+ } catch (TServiceException e) {
+ abort();
+ throw new IOException(e.getMessage(), e);
+ } catch (Throwable t) {
+ throw new IOException(t.getMessage(), t);
+ }
+ if (!ResultCode.OK.name().equals(response.getResultCode()) || response.getErrorMessage() != null) {
+ abort();
+ throw new IOException(response.getErrorMessage());
+ }
+ return response;
+ }
+ }.withRetries();
+
+ if (response != null && response.getQueryId() != null) {
+ return this.getQueryResultAndWait(response.getQueryId(), response);
+ } else {
+ return createNullResultSet(QueryIdFactory.NULL_QUERY_ID.toString());
+ }
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ throw new IOException(e.getMessage(), e);
+ }
+ }
+
+ public ResultSet createNullResultSet(String queryId) throws IOException {
+ TGetQueryStatusResponse emptyQueryStatus = new TGetQueryStatusResponse();
+ emptyQueryStatus.setResultCode(ResultCode.OK.name());
+ emptyQueryStatus.setState(QueryState.QUERY_SUCCEEDED.name());
+ emptyQueryStatus.setQueryId(queryId);
+
+ TQueryResult emptyQueryResult = new TQueryResult();
+
+ emptyQueryResult.setRows(Collections.<ByteBuffer>emptyList());
+ return new TajoThriftResultSet(this, queryId, emptyQueryResult);
+ }
+
+ public ResultSet getQueryResultAndWait(String queryId, TGetQueryStatusResponse queryResponse) throws Exception {
+ if (queryResponse.getQueryResult() != null) {
+ //select 1+1
+ TQueryResult queryResult = queryResponse.getQueryResult();
+ ResultSet resultSet = new TajoThriftMemoryResultSet(this, queryId,
+ TajoThriftUtil.convertSchema(queryResult.getSchema()),
+ queryResult.getRows(), queryResult.getRows() == null ? 0 : queryResult.getRows().size());
+ return resultSet;
+ }
+
+ TGetQueryStatusResponse status = getQueryStatus(queryId);
+ while(status != null && TajoThriftUtil.isQueryRunnning(status.getState())) {
+ status = getQueryStatus(queryId);
+ try {
+ //TODO use thread
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ if (QueryState.QUERY_SUCCEEDED.name().equals(status.getState())) {
+ if (status.isHasResult()) {
+ //select * from lineitem
--- End diff --
This comment is not for debugging. It shows a example.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---