You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by hy...@apache.org on 2014/10/28 06:18:56 UTC
[1/3] TAJO-1140: Separate TajoClient into fine grained parts.
Repository: tajo
Updated Branches:
refs/heads/master 8741e6829 -> f80beaf64
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/JdbcConnection.java
----------------------------------------------------------------------
diff --git a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/JdbcConnection.java b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/JdbcConnection.java
new file mode 100644
index 0000000..4250da4
--- /dev/null
+++ b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/JdbcConnection.java
@@ -0,0 +1,444 @@
+/**
+ * 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.jdbc;
+
+import com.google.protobuf.ServiceException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.TajoConstants;
+import org.apache.tajo.client.CatalogAdminClient;
+import org.apache.tajo.client.QueryClient;
+import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.TajoClientImpl;
+import org.apache.tajo.conf.TajoConf;
+import org.jboss.netty.handler.codec.http.QueryStringDecoder;
+
+import java.io.IOException;
+import java.net.URI;
+import java.sql.*;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class JdbcConnection implements Connection {
+ private static Log LOG = LogFactory.getLog(JdbcConnection.class);
+
+ private final TajoClient tajoClient;
+ private final AtomicBoolean closed = new AtomicBoolean(true);
+ private final String rawURI;
+ private final Properties properties;
+
+ private final URI uri;
+ private final String hostName;
+ private final int port;
+ private final String databaseName;
+ @SuppressWarnings("unused")
+ /** it will be used soon. */
+ private final Map<String, List<String>> params;
+
+ public JdbcConnection(String rawURI, Properties properties) throws SQLException {
+ this.rawURI = rawURI;
+ this.properties = properties;
+
+ try {
+ if (!rawURI.startsWith(TajoDriver.TAJO_JDBC_URL_PREFIX)) {
+ throw new SQLException("Invalid URL: " + rawURI, "TAJO-001");
+ }
+
+ // URI form: jdbc:tajo://hostname:port/databasename
+ int startIdx = rawURI.indexOf(":");
+ if (startIdx < 0) {
+ throw new SQLException("Invalid URL: " + rawURI, "TAJO-001");
+ }
+
+ String uri = rawURI.substring(startIdx+1, rawURI.length());
+ try {
+ this.uri = URI.create(uri);
+ } catch (IllegalArgumentException iae) {
+ throw new SQLException("Invalid URL: " + rawURI, "TAJO-001");
+ }
+
+ hostName = this.uri.getHost();
+ if(hostName == null) {
+ throw new SQLException("Invalid JDBC URI: " + rawURI, "TAJO-001");
+ }
+ if (this.uri.getPort() < 1) {
+ port = 26002;
+ } else {
+ port = this.uri.getPort();
+ }
+
+ if (this.uri.getPath() == null || this.uri.getPath().equalsIgnoreCase("")) { // if no database is given, set default.
+ databaseName = TajoConstants.DEFAULT_DATABASE_NAME;
+ } else {
+ // getPath() will return '/database'.
+ databaseName = this.uri.getPath().split("/")[1];
+ }
+
+ params = new QueryStringDecoder(rawURI).getParameters();
+ } catch (SQLException se) {
+ throw se;
+ } catch (Throwable t) { // for unexpected exceptions like ArrayIndexOutOfBoundsException.
+ throw new SQLException("Invalid JDBC URI: " + rawURI, "TAJO-001");
+ }
+
+ TajoConf tajoConf = new TajoConf();
+ if(properties != null) {
+ for(Map.Entry<Object, Object> entry: properties.entrySet()) {
+ tajoConf.set(entry.getKey().toString(), entry.getValue().toString());
+ }
+ }
+
+ try {
+ tajoClient = new TajoClientImpl(hostName, port, databaseName);
+ } catch (Exception e) {
+ throw new SQLException("Cannot create TajoClient instance:" + e.getMessage(), "TAJO-002");
+ }
+ closed.set(false);
+ }
+
+ public String getUri() {
+ return this.rawURI;
+ }
+
+ public QueryClient getQueryClient() {
+ return tajoClient;
+ }
+
+ public CatalogAdminClient getCatalogAdminClient() {
+ return tajoClient;
+ }
+
+ @Override
+ public void clearWarnings() throws SQLException {
+ }
+
+ @Override
+ public void close() throws SQLException {
+ if(!closed.get()) {
+ if(tajoClient != null) {
+ tajoClient.close();
+ }
+
+ closed.set(true);
+ }
+ }
+
+ @Override
+ public void commit() throws SQLException {
+ throw new SQLFeatureNotSupportedException("commit");
+ }
+
+ @Override
+ public Array createArrayOf(String arg0, Object[] arg1) throws SQLException {
+ throw new SQLFeatureNotSupportedException("createArrayOf");
+ }
+
+ @Override
+ public Blob createBlob() throws SQLException {
+ throw new SQLFeatureNotSupportedException("createBlob");
+ }
+
+ @Override
+ public Clob createClob() throws SQLException {
+ throw new SQLFeatureNotSupportedException("createClob");
+ }
+
+ @Override
+ public NClob createNClob() throws SQLException {
+ throw new SQLFeatureNotSupportedException("createNClob");
+ }
+
+ @Override
+ public SQLXML createSQLXML() throws SQLException {
+ throw new SQLFeatureNotSupportedException("createSQLXML");
+ }
+
+ @Override
+ public Statement createStatement() throws SQLException {
+ if (isClosed()) {
+ throw new SQLException("Can't create Statement, connection is closed");
+ }
+ return new TajoStatement(this, tajoClient);
+ }
+
+ @Override
+ public Statement createStatement(int resultSetType, int resultSetConcurrency)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("createStatement");
+ }
+
+ @Override
+ public Statement createStatement(int resultSetType, int resultSetConcurrency,
+ int resultSetHoldability) throws SQLException {
+ throw new SQLFeatureNotSupportedException("createStatement");
+ }
+
+ @Override
+ public Struct createStruct(String typeName, Object[] attributes)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("createStruct");
+ }
+
+ @Override
+ public boolean getAutoCommit() throws SQLException {
+ return true;
+ }
+
+ @Override
+ public String getCatalog() throws SQLException {
+ try {
+ return tajoClient.getCurrentDatabase();
+ } catch (ServiceException e) {
+ throw new SQLException(e);
+ }
+ }
+
+ @Override
+ public Properties getClientInfo() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getClientInfo");
+ }
+
+ @Override
+ public String getClientInfo(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getClientInfo");
+ }
+
+ @Override
+ public int getHoldability() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getHoldability");
+ }
+
+ @Override
+ public DatabaseMetaData getMetaData() throws SQLException {
+ return new TajoDatabaseMetaData(this);
+ }
+
+ @Override
+ public int getTransactionIsolation() throws SQLException {
+ return Connection.TRANSACTION_NONE;
+ }
+
+ @Override
+ public Map<String, Class<?>> getTypeMap() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getTypeMap");
+ }
+
+ @Override
+ public SQLWarning getWarnings() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getWarnings");
+ }
+
+ @Override
+ public boolean isClosed() throws SQLException {
+ return closed.get();
+ }
+
+ @Override
+ public boolean isReadOnly() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean isValid(int timeout) throws SQLException {
+ try {
+ if (tajoClient.isConnected()) {
+ ResultSet resultSet = tajoClient.executeQueryAndGetResult("SELECT 1;");
+ boolean next = resultSet.next();
+ boolean valid = next && resultSet.getLong(1) == 1;
+ resultSet.close();
+ return valid;
+ } else {
+ return false;
+ }
+ } catch (ServiceException e) {
+ LOG.error("TajoMaster is not available.");
+ return false;
+ } catch (IOException e) {
+ LOG.error("JDBC connection is not valid.");
+ return false;
+ }
+ }
+
+ @Override
+ public String nativeSQL(String sql) throws SQLException {
+ throw new SQLFeatureNotSupportedException("nativeSQL");
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql) throws SQLException {
+ throw new SQLFeatureNotSupportedException("prepareCall");
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql, int resultSetType,
+ int resultSetConcurrency) throws SQLException {
+ throw new SQLFeatureNotSupportedException("prepareCall");
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql, int resultSetType,
+ int resultSetConcurrency, int resultSetHoldability) throws SQLException {
+ throw new SQLFeatureNotSupportedException("prepareCall");
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql) throws SQLException {
+ return new TajoPreparedStatement(this, tajoClient, sql);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys)
+ throws SQLException {
+ return new TajoPreparedStatement(this, tajoClient, sql);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int[] columnIndexes)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("prepareStatement");
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, String[] columnNames)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("prepareStatement");
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int resultSetType,
+ int resultSetConcurrency) throws SQLException {
+ return new TajoPreparedStatement(this, tajoClient, sql);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int resultSetType,
+ int resultSetConcurrency, int resultSetHoldability) throws SQLException {
+ throw new SQLFeatureNotSupportedException("prepareStatement");
+ }
+
+ @Override
+ public void releaseSavepoint(Savepoint savepoint) throws SQLException {
+ throw new SQLFeatureNotSupportedException("releaseSavepoint");
+ }
+
+ @Override
+ public void rollback() throws SQLException {
+ throw new SQLFeatureNotSupportedException("rollback");
+ }
+
+ @Override
+ public void rollback(Savepoint savepoint) throws SQLException {
+ throw new SQLFeatureNotSupportedException("rollback");
+ }
+
+ @Override
+ public void setAutoCommit(boolean autoCommit) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setAutoCommit");
+ }
+
+ @Override
+ public void setCatalog(String catalog) throws SQLException {
+ try {
+ tajoClient.selectDatabase(catalog);
+ } catch (ServiceException e) {
+ throw new SQLException(e);
+ }
+ }
+
+ @Override
+ public void setClientInfo(Properties properties)
+ throws SQLClientInfoException {
+ throw new UnsupportedOperationException("setClientInfo");
+ }
+
+ @Override
+ public void setClientInfo(String name, String value)
+ throws SQLClientInfoException {
+ throw new UnsupportedOperationException("setClientInfo");
+ }
+
+ @Override
+ public void setHoldability(int holdability) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setHoldability");
+ }
+
+ @Override
+ public void setReadOnly(boolean readOnly) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setReadOnly");
+ }
+
+ @Override
+ public Savepoint setSavepoint() throws SQLException {
+ throw new SQLFeatureNotSupportedException("setSavepoint");
+ }
+
+ @Override
+ public Savepoint setSavepoint(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setSavepoint");
+ }
+
+ @Override
+ public void setTransactionIsolation(int level) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setTransactionIsolation");
+ }
+
+ @Override
+ public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setTypeMap");
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> tClass) throws SQLException {
+ if (isWrapperFor(tClass)) {
+ return (T) this;
+ }
+ throw new SQLException("No wrapper for " + tClass);
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> tClass) throws SQLException {
+ return tClass.isInstance(this);
+ }
+
+ public void abort(Executor executor) throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("abort is not supported");
+ }
+
+ public int getNetworkTimeout() throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("getNetworkTimeout is not supported");
+ }
+
+ public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("setNetworkTimeout not supported");
+ }
+
+ public String getSchema() throws SQLException {
+ return TajoConstants.DEFAULT_SCHEMA_NAME;
+ }
+
+ public void setSchema(String schema) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setSchema() is not supported yet");
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoConnection.java
----------------------------------------------------------------------
diff --git a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoConnection.java b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoConnection.java
deleted file mode 100644
index 1a2c6c5..0000000
--- a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoConnection.java
+++ /dev/null
@@ -1,437 +0,0 @@
-/**
- * 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.jdbc;
-
-import com.google.protobuf.ServiceException;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.tajo.TajoConstants;
-import org.apache.tajo.client.TajoClient;
-import org.apache.tajo.conf.TajoConf;
-import org.jboss.netty.handler.codec.http.QueryStringDecoder;
-
-import java.io.IOException;
-import java.net.URI;
-import java.sql.*;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.Executor;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-public class TajoConnection implements Connection {
- private static Log LOG = LogFactory.getLog(TajoConnection.class);
-
- private final TajoClient tajoClient;
- private final AtomicBoolean closed = new AtomicBoolean(true);
- private final String rawURI;
- private final Properties properties;
-
- private final URI uri;
- private final String hostName;
- private final int port;
- private final String databaseName;
- @SuppressWarnings("unused")
- /** it will be used soon. */
- private final Map<String, List<String>> params;
-
- public TajoConnection(String rawURI, Properties properties) throws SQLException {
- this.rawURI = rawURI;
- this.properties = properties;
-
- try {
- if (!rawURI.startsWith(TajoDriver.TAJO_JDBC_URL_PREFIX)) {
- throw new SQLException("Invalid URL: " + rawURI, "TAJO-001");
- }
-
- // URI form: jdbc:tajo://hostname:port/databasename
- int startIdx = rawURI.indexOf(":");
- if (startIdx < 0) {
- throw new SQLException("Invalid URL: " + rawURI, "TAJO-001");
- }
-
- String uri = rawURI.substring(startIdx+1, rawURI.length());
- try {
- this.uri = URI.create(uri);
- } catch (IllegalArgumentException iae) {
- throw new SQLException("Invalid URL: " + rawURI, "TAJO-001");
- }
-
- hostName = this.uri.getHost();
- if(hostName == null) {
- throw new SQLException("Invalid JDBC URI: " + rawURI, "TAJO-001");
- }
- if (this.uri.getPort() < 1) {
- port = 26002;
- } else {
- port = this.uri.getPort();
- }
-
- if (this.uri.getPath() == null || this.uri.getPath().equalsIgnoreCase("")) { // if no database is given, set default.
- databaseName = TajoConstants.DEFAULT_DATABASE_NAME;
- } else {
- // getPath() will return '/database'.
- databaseName = this.uri.getPath().split("/")[1];
- }
-
- params = new QueryStringDecoder(rawURI).getParameters();
- } catch (SQLException se) {
- throw se;
- } catch (Throwable t) { // for unexpected exceptions like ArrayIndexOutOfBoundsException.
- throw new SQLException("Invalid JDBC URI: " + rawURI, "TAJO-001");
- }
-
- TajoConf tajoConf = new TajoConf();
- if(properties != null) {
- for(Map.Entry<Object, Object> entry: properties.entrySet()) {
- tajoConf.set(entry.getKey().toString(), entry.getValue().toString());
- }
- }
-
- try {
- tajoClient = new TajoClient(hostName, port, databaseName);
- } catch (Exception e) {
- throw new SQLException("Cannot create TajoClient instance:" + e.getMessage(), "TAJO-002");
- }
- closed.set(false);
- }
-
- public String getUri() {
- return this.rawURI;
- }
-
- public TajoClient getTajoClient() {
- return tajoClient;
- }
-
- @Override
- public void clearWarnings() throws SQLException {
- }
-
- @Override
- public void close() throws SQLException {
- if(!closed.get()) {
- if(tajoClient != null) {
- tajoClient.close();
- }
-
- closed.set(true);
- }
- }
-
- @Override
- public void commit() throws SQLException {
- throw new SQLFeatureNotSupportedException("commit");
- }
-
- @Override
- public Array createArrayOf(String arg0, Object[] arg1) throws SQLException {
- throw new SQLFeatureNotSupportedException("createArrayOf");
- }
-
- @Override
- public Blob createBlob() throws SQLException {
- throw new SQLFeatureNotSupportedException("createBlob");
- }
-
- @Override
- public Clob createClob() throws SQLException {
- throw new SQLFeatureNotSupportedException("createClob");
- }
-
- @Override
- public NClob createNClob() throws SQLException {
- throw new SQLFeatureNotSupportedException("createNClob");
- }
-
- @Override
- public SQLXML createSQLXML() throws SQLException {
- throw new SQLFeatureNotSupportedException("createSQLXML");
- }
-
- @Override
- public Statement createStatement() throws SQLException {
- if (isClosed()) {
- throw new SQLException("Can't create Statement, connection is closed");
- }
- return new TajoStatement(this, tajoClient);
- }
-
- @Override
- public Statement createStatement(int resultSetType, int resultSetConcurrency)
- throws SQLException {
- throw new SQLFeatureNotSupportedException("createStatement");
- }
-
- @Override
- public Statement createStatement(int resultSetType, int resultSetConcurrency,
- int resultSetHoldability) throws SQLException {
- throw new SQLFeatureNotSupportedException("createStatement");
- }
-
- @Override
- public Struct createStruct(String typeName, Object[] attributes)
- throws SQLException {
- throw new SQLFeatureNotSupportedException("createStruct");
- }
-
- @Override
- public boolean getAutoCommit() throws SQLException {
- return true;
- }
-
- @Override
- public String getCatalog() throws SQLException {
- try {
- return tajoClient.getCurrentDatabase();
- } catch (ServiceException e) {
- throw new SQLException(e);
- }
- }
-
- @Override
- public Properties getClientInfo() throws SQLException {
- throw new SQLFeatureNotSupportedException("getClientInfo");
- }
-
- @Override
- public String getClientInfo(String name) throws SQLException {
- throw new SQLFeatureNotSupportedException("getClientInfo");
- }
-
- @Override
- public int getHoldability() throws SQLException {
- throw new SQLFeatureNotSupportedException("getHoldability");
- }
-
- @Override
- public DatabaseMetaData getMetaData() throws SQLException {
- return new TajoDatabaseMetaData(this);
- }
-
- @Override
- public int getTransactionIsolation() throws SQLException {
- return Connection.TRANSACTION_NONE;
- }
-
- @Override
- public Map<String, Class<?>> getTypeMap() throws SQLException {
- throw new SQLFeatureNotSupportedException("getTypeMap");
- }
-
- @Override
- public SQLWarning getWarnings() throws SQLException {
- throw new SQLFeatureNotSupportedException("getWarnings");
- }
-
- @Override
- public boolean isClosed() throws SQLException {
- return closed.get();
- }
-
- @Override
- public boolean isReadOnly() throws SQLException {
- return false;
- }
-
- @Override
- public boolean isValid(int timeout) throws SQLException {
- try {
- if (tajoClient.isConnected()) {
- ResultSet resultSet = tajoClient.executeQueryAndGetResult("SELECT 1;");
- boolean next = resultSet.next();
- boolean valid = next && resultSet.getLong(1) == 1;
- resultSet.close();
- return valid;
- } else {
- return false;
- }
- } catch (ServiceException e) {
- LOG.error("TajoMaster is not available.");
- return false;
- } catch (IOException e) {
- LOG.error("JDBC connection is not valid.");
- return false;
- }
- }
-
- @Override
- public String nativeSQL(String sql) throws SQLException {
- throw new SQLFeatureNotSupportedException("nativeSQL");
- }
-
- @Override
- public CallableStatement prepareCall(String sql) throws SQLException {
- throw new SQLFeatureNotSupportedException("prepareCall");
- }
-
- @Override
- public CallableStatement prepareCall(String sql, int resultSetType,
- int resultSetConcurrency) throws SQLException {
- throw new SQLFeatureNotSupportedException("prepareCall");
- }
-
- @Override
- public CallableStatement prepareCall(String sql, int resultSetType,
- int resultSetConcurrency, int resultSetHoldability) throws SQLException {
- throw new SQLFeatureNotSupportedException("prepareCall");
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql) throws SQLException {
- return new TajoPreparedStatement(this, tajoClient, sql);
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys)
- throws SQLException {
- return new TajoPreparedStatement(this, tajoClient, sql);
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql, int[] columnIndexes)
- throws SQLException {
- throw new SQLFeatureNotSupportedException("prepareStatement");
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql, String[] columnNames)
- throws SQLException {
- throw new SQLFeatureNotSupportedException("prepareStatement");
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql, int resultSetType,
- int resultSetConcurrency) throws SQLException {
- return new TajoPreparedStatement(this, tajoClient, sql);
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql, int resultSetType,
- int resultSetConcurrency, int resultSetHoldability) throws SQLException {
- throw new SQLFeatureNotSupportedException("prepareStatement");
- }
-
- @Override
- public void releaseSavepoint(Savepoint savepoint) throws SQLException {
- throw new SQLFeatureNotSupportedException("releaseSavepoint");
- }
-
- @Override
- public void rollback() throws SQLException {
- throw new SQLFeatureNotSupportedException("rollback");
- }
-
- @Override
- public void rollback(Savepoint savepoint) throws SQLException {
- throw new SQLFeatureNotSupportedException("rollback");
- }
-
- @Override
- public void setAutoCommit(boolean autoCommit) throws SQLException {
- throw new SQLFeatureNotSupportedException("setAutoCommit");
- }
-
- @Override
- public void setCatalog(String catalog) throws SQLException {
- try {
- tajoClient.selectDatabase(catalog);
- } catch (ServiceException e) {
- throw new SQLException(e);
- }
- }
-
- @Override
- public void setClientInfo(Properties properties)
- throws SQLClientInfoException {
- throw new UnsupportedOperationException("setClientInfo");
- }
-
- @Override
- public void setClientInfo(String name, String value)
- throws SQLClientInfoException {
- throw new UnsupportedOperationException("setClientInfo");
- }
-
- @Override
- public void setHoldability(int holdability) throws SQLException {
- throw new SQLFeatureNotSupportedException("setHoldability");
- }
-
- @Override
- public void setReadOnly(boolean readOnly) throws SQLException {
- throw new SQLFeatureNotSupportedException("setReadOnly");
- }
-
- @Override
- public Savepoint setSavepoint() throws SQLException {
- throw new SQLFeatureNotSupportedException("setSavepoint");
- }
-
- @Override
- public Savepoint setSavepoint(String name) throws SQLException {
- throw new SQLFeatureNotSupportedException("setSavepoint");
- }
-
- @Override
- public void setTransactionIsolation(int level) throws SQLException {
- throw new SQLFeatureNotSupportedException("setTransactionIsolation");
- }
-
- @Override
- public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
- throw new SQLFeatureNotSupportedException("setTypeMap");
- }
-
- @Override
- public <T> T unwrap(Class<T> tClass) throws SQLException {
- if (isWrapperFor(tClass)) {
- return (T) this;
- }
- throw new SQLException("No wrapper for " + tClass);
- }
-
- @Override
- public boolean isWrapperFor(Class<?> tClass) throws SQLException {
- return tClass.isInstance(this);
- }
-
- public void abort(Executor executor) throws SQLException {
- // JDK 1.7
- throw new SQLFeatureNotSupportedException("abort is not supported");
- }
-
- public int getNetworkTimeout() throws SQLException {
- // JDK 1.7
- throw new SQLFeatureNotSupportedException("getNetworkTimeout is not supported");
- }
-
- public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
- // JDK 1.7
- throw new SQLFeatureNotSupportedException("setNetworkTimeout not supported");
- }
-
- public String getSchema() throws SQLException {
- return TajoConstants.DEFAULT_SCHEMA_NAME;
- }
-
- public void setSchema(String schema) throws SQLException {
- throw new SQLFeatureNotSupportedException("setSchema() is not supported yet");
- }
-}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoDatabaseMetaData.java b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoDatabaseMetaData.java
index 0ceb589..150e9bf 100644
--- a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoDatabaseMetaData.java
+++ b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoDatabaseMetaData.java
@@ -21,15 +21,13 @@ import com.google.common.collect.Lists;
import com.google.protobuf.ServiceException;
import org.apache.tajo.annotation.Nullable;
import org.apache.tajo.catalog.*;
+import org.apache.tajo.client.CatalogAdminClient;
import org.apache.tajo.client.ResultSetUtil;
-import org.apache.tajo.client.TajoClient;
import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.common.type.TajoTypeUtil;
import org.apache.tajo.datum.Datum;
import org.apache.tajo.datum.NullDatum;
import org.apache.tajo.datum.TextDatum;
-import org.apache.tajo.storage.Tuple;
-import org.apache.tajo.storage.VTuple;
import org.apache.tajo.util.VersionInfo;
import java.sql.*;
@@ -49,9 +47,9 @@ public class TajoDatabaseMetaData implements DatabaseMetaData {
"radians,round,sign,sin,sqrt,tan";
private static final String STRING_FUNCTIONS = "ascii,chr,concat,left,length,ltrim,repeat,rtrim,substring";
- private final TajoConnection conn;
+ private final JdbcConnection conn;
- public TajoDatabaseMetaData(TajoConnection conn) {
+ public TajoDatabaseMetaData(JdbcConnection conn) {
this.conn = conn;
}
@@ -375,15 +373,15 @@ public class TajoDatabaseMetaData implements DatabaseMetaData {
}
try {
- TajoClient tajoClient = conn.getTajoClient();
+ CatalogAdminClient catalogAdmin = conn.getCatalogAdminClient();
// if catalog is null, all databases are targets.
if (targetCatalogs.isEmpty()) {
- targetCatalogs.addAll(tajoClient.getAllDatabaseNames());
+ targetCatalogs.addAll(catalogAdmin.getAllDatabaseNames());
}
for (String databaseName : targetCatalogs) {
- List<String> tableNames = tajoClient.getTableList(databaseName);
+ List<String> tableNames = catalogAdmin.getTableList(databaseName);
for (String eachTableName: tableNames) {
if (eachTableName.matches(regtableNamePattern)) {
MetaDataTuple tuple = new MetaDataTuple(5);
@@ -424,7 +422,7 @@ public class TajoDatabaseMetaData implements DatabaseMetaData {
public ResultSet getSchemas() throws SQLException {
String databaseName;
try {
- databaseName = conn.getTajoClient().getCurrentDatabase();
+ databaseName = conn.getQueryClient().getCurrentDatabase();
} catch (ServiceException e) {
throw new SQLException(e);
}
@@ -443,7 +441,7 @@ public class TajoDatabaseMetaData implements DatabaseMetaData {
public ResultSet getCatalogs() throws SQLException {
Collection<String> databaseNames;
try {
- databaseNames = conn.getTajoClient().getAllDatabaseNames();
+ databaseNames = conn.getCatalogAdminClient().getAllDatabaseNames();
} catch (ServiceException e) {
throw new SQLException(e);
}
@@ -501,16 +499,17 @@ public class TajoDatabaseMetaData implements DatabaseMetaData {
List<MetaDataTuple> columns = new ArrayList<MetaDataTuple>();
try {
if (targetCatalogs.isEmpty()) {
- targetCatalogs.addAll(conn.getTajoClient().getAllDatabaseNames());
+ targetCatalogs.addAll(conn.getCatalogAdminClient().getAllDatabaseNames());
}
for (String databaseName : targetCatalogs) {
String regtableNamePattern = convertPattern(tableNamePattern == null ? null : tableNamePattern);
String regcolumnNamePattern = convertPattern(columnNamePattern == null ? null : columnNamePattern);
- List<String> tables = conn.getTajoClient().getTableList(databaseName);
+ List<String> tables = conn.getCatalogAdminClient().getTableList(databaseName);
for (String table: tables) {
if (table.matches(regtableNamePattern)) {
- TableDesc tableDesc = conn.getTajoClient().getTableDesc(CatalogUtil.buildFQName(databaseName, table));
+ TableDesc tableDesc = conn.getCatalogAdminClient().getTableDesc(
+ CatalogUtil.buildFQName(databaseName, table));
int pos = 0;
for (Column column: tableDesc.getLogicalSchema().getColumns()) {
@@ -766,7 +765,7 @@ public class TajoDatabaseMetaData implements DatabaseMetaData {
public ResultSet getSchemas(String catalog, String schemaPattern) throws SQLException {
String databaseName;
try {
- databaseName = conn.getTajoClient().getCurrentDatabase();
+ databaseName = conn.getQueryClient().getCurrentDatabase();
} catch (ServiceException e) {
throw new SQLException(e);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoDriver.java
----------------------------------------------------------------------
diff --git a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoDriver.java b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoDriver.java
index cfc5299..73c3060 100644
--- a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoDriver.java
+++ b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoDriver.java
@@ -52,7 +52,7 @@ public class TajoDriver implements Driver, Closeable {
@Override
public Connection connect(String url, Properties properties) throws SQLException {
- return acceptsURL(url) ? new TajoConnection(url, properties) : null;
+ return acceptsURL(url) ? new JdbcConnection(url, properties) : null;
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoPreparedStatement.java
----------------------------------------------------------------------
diff --git a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoPreparedStatement.java b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoPreparedStatement.java
index d4c89c0..fa3df98 100644
--- a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoPreparedStatement.java
+++ b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoPreparedStatement.java
@@ -31,7 +31,7 @@ import java.util.HashMap;
*
*/
public class TajoPreparedStatement implements PreparedStatement {
- private TajoConnection conn;
+ private JdbcConnection conn;
private final String sql;
private TajoClient tajoClient;
/**
@@ -66,7 +66,7 @@ public class TajoPreparedStatement implements PreparedStatement {
/**
*
*/
- public TajoPreparedStatement(TajoConnection conn,
+ public TajoPreparedStatement(JdbcConnection conn,
TajoClient tajoClient,
String sql) {
this.conn = conn;
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoStatement.java
----------------------------------------------------------------------
diff --git a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoStatement.java b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoStatement.java
index 69aa68e..8308211 100644
--- a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoStatement.java
+++ b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/TajoStatement.java
@@ -21,13 +21,12 @@ import com.google.common.collect.Lists;
import com.google.protobuf.ServiceException;
import org.apache.tajo.client.TajoClient;
-import java.io.IOException;
import java.sql.*;
import java.util.HashMap;
import java.util.Map;
public class TajoStatement implements Statement {
- private TajoConnection conn;
+ private JdbcConnection conn;
private TajoClient tajoClient;
private int fetchSize = 200;
@@ -50,7 +49,7 @@ public class TajoStatement implements Statement {
*/
private boolean isClosed = false;
- public TajoStatement(TajoConnection conn, TajoClient tajoClient) {
+ public TajoStatement(JdbcConnection conn, TajoClient tajoClient) {
this.conn = conn;
this.tajoClient = tajoClient;
}
[2/3] TAJO-1140: Separate TajoClient into fine grained parts.
Posted by hy...@apache.org.
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/TajoClient.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/TajoClient.java b/tajo-client/src/main/java/org/apache/tajo/client/TajoClient.java
index 82bd855..376f63f 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/TajoClient.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/TajoClient.java
@@ -18,1031 +18,10 @@
package org.apache.tajo.client;
-import com.google.protobuf.ServiceException;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.tajo.QueryId;
-import org.apache.tajo.QueryIdFactory;
-import org.apache.tajo.SessionVars;
-import org.apache.tajo.TajoIdProtos;
-import org.apache.tajo.TajoProtos.QueryState;
-import org.apache.tajo.annotation.Nullable;
import org.apache.tajo.annotation.ThreadSafe;
-import org.apache.tajo.catalog.CatalogUtil;
-import org.apache.tajo.catalog.Schema;
-import org.apache.tajo.catalog.TableDesc;
-import org.apache.tajo.catalog.TableMeta;
-import org.apache.tajo.catalog.partition.PartitionDesc;
-import org.apache.tajo.catalog.partition.PartitionMethodDesc;
-import org.apache.tajo.catalog.proto.CatalogProtos;
-import org.apache.tajo.catalog.proto.CatalogProtos.PartitionMethodProto;
-import org.apache.tajo.cli.InvalidClientSessionException;
-import org.apache.tajo.conf.TajoConf;
-import org.apache.tajo.conf.TajoConf.ConfVars;
-import org.apache.tajo.ipc.ClientProtos;
-import org.apache.tajo.ipc.ClientProtos.*;
-import org.apache.tajo.ipc.QueryMasterClientProtocol;
-import org.apache.tajo.ipc.QueryMasterClientProtocol.QueryMasterClientProtocolService;
-import org.apache.tajo.ipc.TajoMasterClientProtocol;
-import org.apache.tajo.ipc.TajoMasterClientProtocol.TajoMasterClientProtocolService;
-import org.apache.tajo.jdbc.FetchResultSet;
-import org.apache.tajo.jdbc.SQLStates;
-import org.apache.tajo.jdbc.TajoMemoryResultSet;
-import org.apache.tajo.jdbc.TajoResultSet;
-import org.apache.tajo.rpc.NettyClientBase;
-import org.apache.tajo.rpc.RpcConnectionPool;
-import org.apache.tajo.rpc.ServerCallable;
-import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.KeyValueProto;
-import org.apache.tajo.util.HAServiceUtil;
-import org.apache.tajo.util.KeyValueSet;
-import org.apache.tajo.util.NetUtils;
import java.io.Closeable;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import static org.apache.tajo.ipc.ClientProtos.SerializedResultSet;
@ThreadSafe
-public class TajoClient implements Closeable {
- public static final int UNKNOWN_ROW_NUMBER = -1;
-
- private final Log LOG = LogFactory.getLog(TajoClient.class);
-
- private final TajoConf conf;
-
- private final Map<QueryId, InetSocketAddress> queryMasterMap = new ConcurrentHashMap<QueryId, InetSocketAddress>();
-
- private final InetSocketAddress tajoMasterAddr;
-
- private final RpcConnectionPool connPool;
-
- private final String baseDatabase;
-
- private final UserGroupInformation userInfo;
-
- private volatile TajoIdProtos.SessionIdProto sessionId;
-
- private AtomicBoolean closed = new AtomicBoolean(false);
-
- public TajoClient(TajoConf conf) throws IOException {
- this(conf, NetUtils.createSocketAddr(conf.getVar(ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS)), null);
- }
-
- public TajoClient(TajoConf conf, @Nullable String baseDatabase) throws IOException {
- this(conf, NetUtils.createSocketAddr(conf.getVar(ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS)), baseDatabase);
- }
-
- /**
- * Connect to TajoMaster
- *
- * @param conf TajoConf
- * @param addr TajoMaster address
- * @param baseDatabase The base database name. It is case sensitive. If it is null,
- * the 'default' database will be used.
- * @throws IOException
- */
- public TajoClient(TajoConf conf, InetSocketAddress addr, @Nullable String baseDatabase) throws IOException {
- this.conf = conf;
- this.conf.set("tajo.disk.scheduler.report.interval", "0");
- this.tajoMasterAddr = addr;
- int workerNum = conf.getIntVar(TajoConf.ConfVars.RPC_CLIENT_WORKER_THREAD_NUM);
- // Don't share connection pool per client
- connPool = RpcConnectionPool.newPool(conf, getClass().getSimpleName(), workerNum);
- userInfo = UserGroupInformation.getCurrentUser();
- this.baseDatabase = baseDatabase != null ? baseDatabase : null;
- }
-
- public void setSessionId(TajoIdProtos.SessionIdProto sessionId) {
- this.sessionId = sessionId;
- }
-
- public boolean isConnected() {
- if(!closed.get()){
- try {
- return connPool.getConnection(tajoMasterAddr, TajoMasterClientProtocol.class, false).isConnected();
- } catch (Throwable e) {
- return false;
- }
- }
- return false;
- }
-
- public TajoClient(InetSocketAddress addr) throws IOException {
- this(new TajoConf(), addr, null);
- }
-
- public TajoClient(String hostname, int port, String baseDatabase) throws IOException {
- this(new TajoConf(), NetUtils.createSocketAddr(hostname, port), baseDatabase);
- }
-
- public TajoIdProtos.SessionIdProto getSessionId() {
- return sessionId;
- }
-
- private InetSocketAddress getTajoMasterAddr() {
- if (!conf.getBoolVar(TajoConf.ConfVars.TAJO_MASTER_HA_ENABLE)) {
- return tajoMasterAddr;
- } else {
- if (!HAServiceUtil.isMasterAlive(tajoMasterAddr, conf)) {
- return HAServiceUtil.getMasterClientAddress(conf);
- } else {
- return tajoMasterAddr;
- }
- }
- }
-
- public String getBaseDatabase() {
- return baseDatabase;
- }
-
- @Override
- public void close() {
- if(closed.getAndSet(true)){
- return;
- }
-
- // remove session
- try {
- NettyClientBase client = connPool.getConnection(getTajoMasterAddr(), TajoMasterClientProtocol.class, false);
- TajoMasterClientProtocolService.BlockingInterface tajoMaster = client.getStub();
- tajoMaster.removeSession(null, sessionId);
- } catch (Throwable e) {
- }
-
- if(connPool != null) {
- connPool.shutdown();
- }
- queryMasterMap.clear();
- }
-
- public TajoConf getConf() {
- return conf;
- }
-
- public UserGroupInformation getUserInfo() {
- return userInfo;
- }
-
- /**
- * Call to QueryMaster closing query resources
- * @param queryId
- */
- public void closeQuery(final QueryId queryId) {
- if(queryMasterMap.containsKey(queryId)) {
- NettyClientBase qmClient = null;
- try {
- qmClient = connPool.getConnection(queryMasterMap.get(queryId), QueryMasterClientProtocol.class, false);
- QueryMasterClientProtocolService.BlockingInterface queryMasterService = qmClient.getStub();
- queryMasterService.closeQuery(null, queryId.getProto());
- } catch (Exception e) {
- LOG.warn("Fail to close a QueryMaster connection (qid=" + queryId + ", msg=" + e.getMessage() + ")", e);
- } finally {
- connPool.closeConnection(qmClient);
- queryMasterMap.remove(queryId);
- }
- }
- }
-
- public void closeNonForwardQuery(final QueryId queryId) {
- NettyClientBase tmClient = null;
- try {
- tmClient = connPool.getConnection(getTajoMasterAddr(), TajoMasterClientProtocol.class, false);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = tmClient.getStub();
- checkSessionAndGet(tmClient);
-
- QueryIdRequest.Builder builder = QueryIdRequest.newBuilder();
-
- builder.setSessionId(getSessionId());
- builder.setQueryId(queryId.getProto());
- tajoMasterService.closeNonForwardQuery(null, builder.build());
- } catch (Exception e) {
- LOG.warn("Fail to close a TajoMaster connection (qid=" + queryId + ", msg=" + e.getMessage() + ")", e);
- } finally {
- connPool.closeConnection(tmClient);
- }
- }
-
- private void checkSessionAndGet(NettyClientBase client) throws ServiceException {
- if (sessionId == null) {
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- CreateSessionRequest.Builder builder = CreateSessionRequest.newBuilder();
- builder.setUsername(userInfo.getUserName()).build();
- if (baseDatabase != null) {
- builder.setBaseDatabaseName(baseDatabase);
- }
- CreateSessionResponse response = tajoMasterService.createSession(null, builder.build());
- if (response.getState() == CreateSessionResponse.ResultState.SUCCESS) {
- sessionId = response.getSessionId();
- if (LOG.isDebugEnabled()) {
- LOG.debug(String.format("Got session %s as a user '%s'.", sessionId.getId(), userInfo.getUserName()));
- }
- } else {
- throw new InvalidClientSessionException(response.getMessage());
- }
- }
- }
-
- private SessionedStringProto convertSessionedString(String str) {
- SessionedStringProto.Builder builder = SessionedStringProto.newBuilder();
- builder.setSessionId(sessionId);
- builder.setValue(str);
- return builder.build();
- }
-
- public String getCurrentDatabase() throws ServiceException {
- return new ServerCallable<String>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
-
- public String call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- return tajoMasterService.getCurrentDatabase(null, sessionId).getValue();
- }
- }.withRetries();
- }
-
- public Boolean selectDatabase(final String databaseName) throws ServiceException {
- return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
-
- public Boolean call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- return tajoMasterService.selectDatabase(null, convertSessionedString(databaseName)).getValue();
- }
- }.withRetries();
- }
-
- public Boolean updateSessionVariables(final Map<String, String> variables) throws ServiceException {
- return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
-
- public Boolean call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- KeyValueSet keyValueSet = new KeyValueSet();
- keyValueSet.putAll(variables);
- UpdateSessionVariableRequest request = UpdateSessionVariableRequest.newBuilder()
- .setSessionId(sessionId)
- .setSetVariables(keyValueSet.getProto()).build();
-
- return tajoMasterService.updateSessionVariables(null, request).getValue();
- }
- }.withRetries();
- }
-
- public Boolean unsetSessionVariables(final List<String> variables) throws ServiceException {
- return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
-
- public Boolean call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- UpdateSessionVariableRequest request = UpdateSessionVariableRequest.newBuilder()
- .setSessionId(sessionId)
- .addAllUnsetVariables(variables).build();
- return tajoMasterService.updateSessionVariables(null, request).getValue();
- }
- }.withRetries();
- }
-
- public String getSessionVariable(final String varname) throws ServiceException {
- return new ServerCallable<String>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
-
- public String call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- return tajoMasterService.getSessionVariable(null, convertSessionedString(varname)).getValue();
- }
- }.withRetries();
- }
-
- public Boolean existSessionVariable(final String varname) throws ServiceException {
- return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
-
- public Boolean call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- return tajoMasterService.existSessionVariable(null, convertSessionedString(varname)).getValue();
- }
- }.withRetries();
- }
-
- public Map<String, String> getAllSessionVariables() throws ServiceException {
- return new ServerCallable<Map<String, String>>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class,
- false, true) {
-
- public Map<String, String> call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- KeyValueSet keyValueSet = new KeyValueSet(tajoMasterService.getAllSessionVariables(null, sessionId));
- return keyValueSet.getAllKeyValus();
- }
- }.withRetries();
- }
-
- /**
- * It submits a query statement and get a response immediately.
- * The response only contains a query id, and submission status.
- * In order to get the result, you should use {@link #getQueryResult(org.apache.tajo.QueryId)}
- * or {@link #getQueryResultAndWait(org.apache.tajo.QueryId)}.
- */
- public SubmitQueryResponse executeQuery(final String sql) throws ServiceException {
- return new ServerCallable<SubmitQueryResponse>(connPool, getTajoMasterAddr(),
- TajoMasterClientProtocol.class, false, true) {
- public SubmitQueryResponse call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- final QueryRequest.Builder builder = QueryRequest.newBuilder();
- builder.setSessionId(sessionId);
- builder.setQuery(sql);
- builder.setIsJson(false);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- return tajoMasterService.submitQuery(null, builder.build());
- }
- }.withRetries();
- }
-
- public SubmitQueryResponse executeQueryWithJson(final String json) throws ServiceException {
- return new ServerCallable<SubmitQueryResponse>(connPool, getTajoMasterAddr(),
- TajoMasterClientProtocol.class, false, true) {
- public SubmitQueryResponse call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- final QueryRequest.Builder builder = QueryRequest.newBuilder();
- builder.setSessionId(sessionId);
- builder.setQuery(json);
- builder.setIsJson(true);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- return tajoMasterService.submitQuery(null, builder.build());
- }
- }.withRetries();
- }
-
- /**
- * It submits a query statement and get a response.
- * The main difference from {@link #executeQuery(String)}
- * is a blocking method. So, this method is wait for
- * the finish of the submitted query.
- *
- * @return If failed, return null.
- */
- public ResultSet executeQueryAndGetResult(final String sql)
- throws ServiceException, IOException {
- SubmitQueryResponse response = executeQuery(sql);
-
- if (response.getResultCode() == ClientProtos.ResultCode.ERROR) {
- throw new ServiceException(response.getErrorTrace());
- }
- QueryId queryId = new QueryId(response.getQueryId());
- if (response.getIsForwarded()) {
- if (queryId.equals(QueryIdFactory.NULL_QUERY_ID)) {
- return this.createNullResultSet(queryId);
- } else {
- return this.getQueryResultAndWait(queryId);
- }
- } else {
- // If a non-forwarded insert into query
- if (queryId.equals(QueryIdFactory.NULL_QUERY_ID) && response.getMaxRowNum() == 0) {
- return this.createNullResultSet(queryId);
- } else {
- if (response.hasResultSet() || response.hasTableDesc()) {
- return createResultSet(this, response);
- } else {
- return this.createNullResultSet(queryId);
- }
- }
- }
- }
-
- public ResultSet executeJsonQueryAndGetResult(final String json) throws ServiceException, IOException {
- SubmitQueryResponse response = executeQueryWithJson(json);
- if (response.getResultCode() == ClientProtos.ResultCode.ERROR) {
- throw new ServiceException(response.getErrorTrace());
- }
- QueryId queryId = new QueryId(response.getQueryId());
- if (response.getIsForwarded()) {
- if (queryId.equals(QueryIdFactory.NULL_QUERY_ID)) {
- return this.createNullResultSet(queryId);
- } else {
- return this.getQueryResultAndWait(queryId);
- }
- } else {
- if (response.hasResultSet() || response.hasTableDesc()) {
- return createResultSet(this, response);
- } else {
- return this.createNullResultSet(queryId);
- }
- }
- }
-
- public QueryStatus getQueryStatus(QueryId queryId) throws ServiceException {
- GetQueryStatusRequest.Builder builder = GetQueryStatusRequest.newBuilder();
- builder.setQueryId(queryId.getProto());
-
- GetQueryStatusResponse res = null;
- if(queryMasterMap.containsKey(queryId)) {
- NettyClientBase qmClient = null;
- try {
- qmClient = connPool.getConnection(queryMasterMap.get(queryId),
- QueryMasterClientProtocol.class, false);
- QueryMasterClientProtocolService.BlockingInterface queryMasterService = qmClient.getStub();
- res = queryMasterService.getQueryStatus(null, builder.build());
- } catch (Exception e) {
- throw new ServiceException(e.getMessage(), e);
- } finally {
- connPool.releaseConnection(qmClient);
- }
- } else {
- NettyClientBase tmClient = null;
- try {
- tmClient = connPool.getConnection(getTajoMasterAddr(), TajoMasterClientProtocol.class, false);
-
- checkSessionAndGet(tmClient);
- builder.setSessionId(sessionId);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = tmClient.getStub();
- res = tajoMasterService.getQueryStatus(null, builder.build());
-
- String queryMasterHost = res.getQueryMasterHost();
- if(queryMasterHost != null && !queryMasterHost.isEmpty()) {
- NettyClientBase qmClient = null;
- try {
- InetSocketAddress qmAddr = NetUtils.createSocketAddr(queryMasterHost, res.getQueryMasterPort());
- qmClient = connPool.getConnection(
- qmAddr, QueryMasterClientProtocol.class, false);
- QueryMasterClientProtocolService.BlockingInterface queryMasterService = qmClient.getStub();
- res = queryMasterService.getQueryStatus(null, builder.build());
-
- queryMasterMap.put(queryId, qmAddr);
- } catch (Exception e) {
- throw new ServiceException(e.getMessage(), e);
- } finally {
- connPool.releaseConnection(qmClient);
- }
- }
- } catch (Exception e) {
- throw new ServiceException(e.getMessage(), e);
- } finally {
- connPool.releaseConnection(tmClient);
- }
- }
- return new QueryStatus(res);
- }
-
- /* query submit */
- public static boolean isInPreNewState(QueryState state) {
- return state == QueryState.QUERY_NOT_ASSIGNED ||
- state == QueryState.QUERY_MASTER_INIT ||
- state == QueryState.QUERY_MASTER_LAUNCHED;
- }
-
- /* query submitted. but is not running */
- public static boolean isInInitState(QueryState state) {
- return state == QueryState.QUERY_NEW || state == QueryState.QUERY_INIT;
- }
-
- /* query started. but is not complete */
- public static boolean isInRunningState(QueryState state) {
- return isInInitState(state) || state == QueryState.QUERY_RUNNING;
- }
-
- /* query complete */
- public static boolean isInCompleteState(QueryState state) {
- return !isInPreNewState(state) && !isInRunningState(state);
- }
-
- public ResultSet getQueryResult(QueryId queryId)
- throws ServiceException, IOException {
- if (queryId.equals(QueryIdFactory.NULL_QUERY_ID)) {
- return createNullResultSet(queryId);
- }
- GetQueryResultResponse response = getResultResponse(queryId);
- TableDesc tableDesc = CatalogUtil.newTableDesc(response.getTableDesc());
- conf.setVar(ConfVars.USERNAME, response.getTajoUserName());
- return new TajoResultSet(this, queryId, conf, tableDesc);
- }
-
- public static ResultSet createResultSet(TajoClient client, QueryId queryId, GetQueryResultResponse response)
- throws IOException {
- TableDesc desc = CatalogUtil.newTableDesc(response.getTableDesc());
- TajoConf conf = new TajoConf(client.getConf());
- conf.setVar(ConfVars.USERNAME, response.getTajoUserName());
- return new TajoResultSet(client, queryId, conf, desc);
- }
-
- public static ResultSet createResultSet(TajoClient client, SubmitQueryResponse response) throws IOException {
- if (response.hasTableDesc()) {
- // non-forward query
- // select * from table1 [limit 10]
- int fetchRowNum = client.getConf().getIntVar(ConfVars.$RESULT_SET_FETCH_ROWNUM);
- if (response.hasSessionVariables()) {
- for (KeyValueProto eachKeyValue: response.getSessionVariables().getKeyvalList()) {
- if (eachKeyValue.getKey().equals(SessionVars.FETCH_ROWNUM.keyname())) {
- fetchRowNum = Integer.parseInt(eachKeyValue.getValue());
- }
- }
- }
- TableDesc tableDesc = new TableDesc(response.getTableDesc());
- return new FetchResultSet(client, tableDesc.getLogicalSchema(), new QueryId(response.getQueryId()), fetchRowNum);
- } else {
- // simple eval query
- // select substr('abc', 1, 2)
- SerializedResultSet serializedResultSet = response.getResultSet();
- return new TajoMemoryResultSet(
- new Schema(serializedResultSet.getSchema()),
- serializedResultSet.getSerializedTuplesList(),
- response.getMaxRowNum());
- }
- }
-
- private ResultSet getQueryResultAndWait(QueryId queryId)
- throws ServiceException, IOException {
- if (queryId.equals(QueryIdFactory.NULL_QUERY_ID)) {
- return createNullResultSet(queryId);
- }
- QueryStatus status = getQueryStatus(queryId);
-
- while(status != null && !isInCompleteState(status.getState())) {
- try {
- Thread.sleep(500);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
-
- status = getQueryStatus(queryId);
- }
-
- if (status.getState() == QueryState.QUERY_SUCCEEDED) {
- if (status.hasResult()) {
- return getQueryResult(queryId);
- } else {
- return createNullResultSet(queryId);
- }
-
- } else {
- LOG.warn("Query (" + status.getQueryId() + ") failed: " + status.getState());
-
- //TODO throw SQLException(?)
- return createNullResultSet(queryId);
- }
- }
-
- public ResultSet createNullResultSet(QueryId queryId) throws IOException {
- return new TajoResultSet(this, queryId);
- }
-
- public GetQueryResultResponse getResultResponse(QueryId queryId) throws ServiceException {
- if (queryId.equals(QueryIdFactory.NULL_QUERY_ID)) {
- return null;
- }
-
- NettyClientBase client = null;
- try {
- InetSocketAddress queryMasterAddr = queryMasterMap.get(queryId);
- if(queryMasterAddr == null) {
- LOG.warn("No Connection to QueryMaster for " + queryId);
- return null;
- }
- client = connPool.getConnection(queryMasterAddr, QueryMasterClientProtocol.class, false);
- QueryMasterClientProtocolService.BlockingInterface queryMasterService = client.getStub();
- GetQueryResultRequest.Builder builder = GetQueryResultRequest.newBuilder();
- builder.setQueryId(queryId.getProto());
- GetQueryResultResponse response = queryMasterService.getQueryResult(null,
- builder.build());
-
- return response;
- } catch (Exception e) {
- throw new ServiceException(e.getMessage(), e);
- } finally {
- connPool.releaseConnection(client);
- }
- }
-
- public TajoMemoryResultSet fetchNextQueryResult(final QueryId queryId, final int fetchRowNum) throws ServiceException {
- try {
- ServerCallable<SerializedResultSet> callable =
- new ServerCallable<SerializedResultSet>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
- public SerializedResultSet call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- GetQueryResultDataRequest.Builder builder = GetQueryResultDataRequest.newBuilder();
- builder.setSessionId(sessionId);
- builder.setQueryId(queryId.getProto());
- builder.setFetchRowNum(fetchRowNum);
- try {
- GetQueryResultDataResponse response = tajoMasterService.getQueryResultData(null, builder.build());
- if (response.getResultCode() == ResultCode.ERROR) {
- throw new ServiceException(response.getErrorTrace());
- }
-
- return response.getResultSet();
- } catch (ServiceException e) {
- abort();
- throw e;
- } catch (Throwable t) {
- throw new ServiceException(t.getMessage(), t);
- }
- }
- };
-
- SerializedResultSet serializedResultSet = callable.withRetries();
-
- return new TajoMemoryResultSet(
- new Schema(serializedResultSet.getSchema()),
- serializedResultSet.getSerializedTuplesList(),
- serializedResultSet.getSerializedTuplesCount());
- } catch (Exception e) {
- throw new ServiceException(e.getMessage(), e);
- }
- }
-
- public boolean updateQuery(final String sql) throws ServiceException {
- return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(),
- TajoMasterClientProtocol.class, false, true) {
- public Boolean call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- QueryRequest.Builder builder = QueryRequest.newBuilder();
- builder.setSessionId(sessionId);
- builder.setQuery(sql);
- builder.setIsJson(false);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- UpdateQueryResponse response = tajoMasterService.updateQuery(null, builder.build());
- if (response.getResultCode() == ResultCode.OK) {
- return true;
- } else {
- if (response.hasErrorMessage()) {
- System.err.println("ERROR: " + response.getErrorMessage());
- }
- return false;
- }
- }
- }.withRetries();
- }
-
- public boolean updateQueryWithJson(final String json) throws ServiceException {
- return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(),
- TajoMasterClientProtocol.class, false, true) {
- public Boolean call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- QueryRequest.Builder builder = QueryRequest.newBuilder();
- builder.setSessionId(sessionId);
- builder.setQuery(json);
- builder.setIsJson(true);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- UpdateQueryResponse response = tajoMasterService.updateQuery(null, builder.build());
- if (response.getResultCode() == ResultCode.OK) {
- return true;
- } else {
- if (response.hasErrorMessage()) {
- System.err.println("ERROR: " + response.getErrorMessage());
- }
- return false;
- }
- }
- }.withRetries();
- }
-
- /**
- * Create a database.
- *
- * @param databaseName The database name to be created. This name is case sensitive.
- * @return True if created successfully.
- * @throws ServiceException
- */
- public boolean createDatabase(final String databaseName) throws ServiceException {
- return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
- public Boolean call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- return tajoMasterService.createDatabase(null, convertSessionedString(databaseName)).getValue();
- }
- }.withRetries();
- }
-
- /**
- * Does the database exist?
- *
- * @param databaseName The database name to be checked. This name is case sensitive.
- * @return True if so.
- * @throws ServiceException
- */
- public boolean existDatabase(final String databaseName) throws ServiceException {
- return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
- public Boolean call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- return tajoMasterService.existDatabase(null, convertSessionedString(databaseName)).getValue();
- }
- }.withRetries();
- }
-
- /**
- * Drop the database
- *
- * @param databaseName The database name to be dropped. This name is case sensitive.
- * @return True if the database is dropped successfully.
- * @throws ServiceException
- */
- public boolean dropDatabase(final String databaseName) throws ServiceException {
- return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
- public Boolean call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- return tajoMasterService.dropDatabase(null, convertSessionedString(databaseName)).getValue();
- }
- }.withRetries();
- }
-
- public List<String> getAllDatabaseNames() throws ServiceException {
- return new ServerCallable<List<String>>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
- public List<String> call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- return tajoMasterService.getAllDatabases(null, sessionId).getValuesList();
- }
- }.withRetries();
- }
-
- /**
- * Does the table exist?
- *
- * @param tableName The table name to be checked. This name is case sensitive.
- * @return True if so.
- */
- public boolean existTable(final String tableName) throws ServiceException {
- return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(),
- TajoMasterClientProtocol.class, false, true) {
- public Boolean call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- return tajoMasterService.existTable(null, convertSessionedString(tableName)).getValue();
- }
- }.withRetries();
- }
-
- /**
- * Create an external table.
- *
- * @param tableName The table name to be created. This name is case sensitive. This name can be qualified or not.
- * If the table name is not qualified, the current database in the session will be used.
- * @param schema The schema
- * @param path The external table location
- * @param meta Table meta
- * @return the created table description.
- * @throws SQLException
- * @throws ServiceException
- */
- public TableDesc createExternalTable(final String tableName, final Schema schema, final Path path,
- final TableMeta meta)
- throws SQLException, ServiceException {
- return createExternalTable(tableName, schema, path, meta, null);
- }
-
- /**
- * Create an external table.
- *
- * @param tableName The table name to be created. This name is case sensitive. This name can be qualified or not.
- * If the table name is not qualified, the current database in the session will be used.
- * @param schema The schema
- * @param path The external table location
- * @param meta Table meta
- * @param partitionMethodDesc Table partition description
- * @return the created table description.
- * @throws SQLException
- * @throws ServiceException
- */
- public TableDesc createExternalTable(final String tableName, final Schema schema, final Path path,
- final TableMeta meta, final PartitionMethodDesc partitionMethodDesc)
- throws SQLException, ServiceException {
- return new ServerCallable<TableDesc>(connPool, getTajoMasterAddr(),
- TajoMasterClientProtocol.class, false, true) {
- public TableDesc call(NettyClientBase client) throws ServiceException, SQLException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
-
- CreateTableRequest.Builder builder = CreateTableRequest.newBuilder();
- builder.setSessionId(sessionId);
- builder.setName(tableName);
- builder.setSchema(schema.getProto());
- builder.setMeta(meta.getProto());
- builder.setPath(path.toUri().toString());
- if (partitionMethodDesc != null) {
- builder.setPartition(partitionMethodDesc.getProto());
- }
- TableResponse res = tajoMasterService.createExternalTable(null, builder.build());
- if (res.getResultCode() == ResultCode.OK) {
- return CatalogUtil.newTableDesc(res.getTableDesc());
- } else {
- throw new SQLException(res.getErrorMessage(), SQLStates.ER_NO_SUCH_TABLE.getState());
- }
- }
- }.withRetries();
- }
-
- /**
- * Drop a table
- *
- * @param tableName The table name to be dropped. This name is case sensitive.
- * @return True if the table is dropped successfully.
- */
- public boolean dropTable(final String tableName) throws ServiceException {
- return dropTable(tableName, false);
- }
-
- /**
- * Drop a table.
- *
- * @param tableName The table name to be dropped. This name is case sensitive.
- * @param purge If purge is true, this call will remove the entry in catalog as well as the table contents.
- * @return True if the table is dropped successfully.
- */
- public boolean dropTable(final String tableName, final boolean purge) throws ServiceException {
- return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(),
- TajoMasterClientProtocol.class, false, true) {
- public Boolean call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
-
- DropTableRequest.Builder builder = DropTableRequest.newBuilder();
- builder.setSessionId(sessionId);
- builder.setName(tableName);
- builder.setPurge(purge);
- return tajoMasterService.dropTable(null, builder.build()).getValue();
- }
- }.withRetries();
-
- }
-
- public List<BriefQueryInfo> getRunningQueryList() throws ServiceException {
- return new ServerCallable<List<BriefQueryInfo>>(connPool, getTajoMasterAddr(),
- TajoMasterClientProtocol.class, false, true) {
- public List<BriefQueryInfo> call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
-
- GetQueryListRequest.Builder builder = GetQueryListRequest.newBuilder();
- builder.setSessionId(sessionId);
- GetQueryListResponse res = tajoMasterService.getRunningQueryList(null, builder.build());
- return res.getQueryListList();
- }
- }.withRetries();
- }
-
- public List<BriefQueryInfo> getFinishedQueryList() throws ServiceException {
- return new ServerCallable<List<BriefQueryInfo>>(connPool, getTajoMasterAddr(),
- TajoMasterClientProtocol.class, false, true) {
- public List<BriefQueryInfo> call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
-
- GetQueryListRequest.Builder builder = GetQueryListRequest.newBuilder();
- builder.setSessionId(sessionId);
- GetQueryListResponse res = tajoMasterService.getFinishedQueryList(null, builder.build());
- return res.getQueryListList();
- }
- }.withRetries();
- }
-
- public List<WorkerResourceInfo> getClusterInfo() throws ServiceException {
- return new ServerCallable<List<WorkerResourceInfo>>(connPool, getTajoMasterAddr(),
- TajoMasterClientProtocol.class, false, true) {
- public List<WorkerResourceInfo> call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
-
- GetClusterInfoRequest.Builder builder = GetClusterInfoRequest.newBuilder();
- builder.setSessionId(sessionId);
- GetClusterInfoResponse res = tajoMasterService.getClusterInfo(null, builder.build());
- return res.getWorkerListList();
- }
- }.withRetries();
- }
-
- /**
- * Get a list of table names.
- *
- * @param databaseName The database name to show all tables. This name is case sensitive.
- * If it is null, this method will show all tables
- * in the current database of this session.
- */
- public List<String> getTableList(@Nullable final String databaseName) throws ServiceException {
- return new ServerCallable<List<String>>(connPool, getTajoMasterAddr(),
- TajoMasterClientProtocol.class, false, true) {
- public List<String> call(NettyClientBase client) throws ServiceException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
-
- GetTableListRequest.Builder builder = GetTableListRequest.newBuilder();
- builder.setSessionId(sessionId);
- if (databaseName != null) {
- builder.setDatabaseName(databaseName);
- }
- GetTableListResponse res = tajoMasterService.getTableList(null, builder.build());
- return res.getTablesList();
- }
- }.withRetries();
- }
-
- /**
- * Get a table description
- *
- * @param tableName The table name to get. This name is case sensitive.
- * @return Table description
- */
- public TableDesc getTableDesc(final String tableName) throws ServiceException {
- return new ServerCallable<TableDesc>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
- public TableDesc call(NettyClientBase client) throws ServiceException, SQLException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
-
- GetTableDescRequest.Builder builder = GetTableDescRequest.newBuilder();
- builder.setSessionId(sessionId);
- builder.setTableName(tableName);
- TableResponse res = tajoMasterService.getTableDesc(null, builder.build());
- if (res.getResultCode() == ResultCode.OK) {
- return CatalogUtil.newTableDesc(res.getTableDesc());
- } else {
- throw new SQLException(res.getErrorMessage(), SQLStates.ER_NO_SUCH_TABLE.getState());
- }
- }
- }.withRetries();
- }
-
- public QueryStatus killQuery(final QueryId queryId)
- throws ServiceException, IOException {
-
- QueryStatus status = getQueryStatus(queryId);
-
- NettyClientBase tmClient = null;
- try {
- /* send a kill to the TM */
- tmClient = connPool.getConnection(getTajoMasterAddr(), TajoMasterClientProtocol.class, false);
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = tmClient.getStub();
-
- checkSessionAndGet(tmClient);
-
- QueryIdRequest.Builder builder = QueryIdRequest.newBuilder();
- builder.setSessionId(sessionId);
- builder.setQueryId(queryId.getProto());
- tajoMasterService.killQuery(null, builder.build());
-
- long currentTimeMillis = System.currentTimeMillis();
- long timeKillIssued = currentTimeMillis;
- while ((currentTimeMillis < timeKillIssued + 10000L)
- && ((status.getState() != QueryState.QUERY_KILLED)
- || (status.getState() == QueryState.QUERY_KILL_WAIT))) {
- try {
- Thread.sleep(100L);
- } catch(InterruptedException ie) {
- break;
- }
- currentTimeMillis = System.currentTimeMillis();
- status = getQueryStatus(queryId);
- }
-
- } catch(Exception e) {
- LOG.debug("Error when checking for application status", e);
- } finally {
- connPool.releaseConnection(tmClient);
- }
- return status;
- }
-
- public List<CatalogProtos.FunctionDescProto> getFunctions(final String functionName) throws ServiceException {
- return new ServerCallable<List<CatalogProtos.FunctionDescProto>>(connPool, getTajoMasterAddr(),
- TajoMasterClientProtocol.class, false, true) {
- public List<CatalogProtos.FunctionDescProto> call(NettyClientBase client) throws ServiceException, SQLException {
- checkSessionAndGet(client);
-
- TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
- String paramFunctionName = functionName == null ? "" : functionName;
- FunctionResponse res = tajoMasterService.getFunctionList(null,convertSessionedString(paramFunctionName));
- if (res.getResultCode() == ResultCode.OK) {
- return res.getFunctionsList();
- } else {
- throw new SQLException(res.getErrorMessage());
- }
- }
- }.withRetries();
- }
+public interface TajoClient extends QueryClient, CatalogAdminClient, Closeable {
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java b/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java
new file mode 100644
index 0000000..75de492
--- /dev/null
+++ b/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java
@@ -0,0 +1,215 @@
+/**
+ * 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.client;
+
+import com.google.protobuf.ServiceException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.QueryId;
+import org.apache.tajo.annotation.Nullable;
+import org.apache.tajo.annotation.ThreadSafe;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.catalog.partition.PartitionMethodDesc;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.conf.TajoConf.ConfVars;
+import org.apache.tajo.ipc.ClientProtos.BriefQueryInfo;
+import org.apache.tajo.ipc.ClientProtos.GetQueryResultResponse;
+import org.apache.tajo.ipc.ClientProtos.SubmitQueryResponse;
+import org.apache.tajo.ipc.ClientProtos.WorkerResourceInfo;
+import org.apache.tajo.jdbc.TajoMemoryResultSet;
+import org.apache.tajo.jdbc.TajoResultSet;
+import org.apache.tajo.util.NetUtils;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.List;
+
+@ThreadSafe
+public class TajoClientImpl extends SessionConnection implements TajoClient, QueryClient, CatalogAdminClient {
+
+ private final Log LOG = LogFactory.getLog(TajoClientImpl.class);
+ QueryClient queryClient;
+ CatalogAdminClient catalogClient;
+
+ public TajoClientImpl(TajoConf conf) throws IOException {
+ this(conf, NetUtils.createSocketAddr(conf.getVar(ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS)), null);
+ }
+
+ public TajoClientImpl(TajoConf conf, @Nullable String baseDatabase) throws IOException {
+ this(conf, NetUtils.createSocketAddr(conf.getVar(ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS)), baseDatabase);
+ }
+
+ public TajoClientImpl(InetSocketAddress addr) throws IOException {
+ this(new TajoConf(), addr, null);
+ }
+
+ /**
+ * Connect to TajoMaster
+ *
+ * @param conf TajoConf
+ * @param addr TajoMaster address
+ * @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 TajoClientImpl(TajoConf conf, InetSocketAddress addr, @Nullable String baseDatabase) throws IOException {
+ super(conf, addr, baseDatabase);
+ this.queryClient = new QueryClientImpl(this);
+ this.catalogClient = new CatalogAdminClientImpl(this);
+ }
+
+ public TajoClientImpl(String hostName, int port, @Nullable String baseDatabase) throws IOException {
+ super(hostName, port, baseDatabase);
+ this.queryClient = new QueryClientImpl(this);
+ this.catalogClient = new CatalogAdminClientImpl(this);
+ }
+
+ /*------------------------------------------------------------------------*/
+ // QueryClient wrappers
+ /*------------------------------------------------------------------------*/
+
+ public void closeQuery(final QueryId queryId) {
+ queryClient.closeQuery(queryId);
+ }
+
+ public void closeNonForwardQuery(final QueryId queryId) {
+ queryClient.closeNonForwardQuery(queryId);
+ }
+
+ public SubmitQueryResponse executeQuery(final String sql) throws ServiceException {
+ return queryClient.executeQuery(sql);
+ }
+
+ public SubmitQueryResponse executeQueryWithJson(final String json) throws ServiceException {
+ return queryClient.executeQueryWithJson(json);
+ }
+
+ public ResultSet executeQueryAndGetResult(final String sql) throws ServiceException, IOException {
+ return queryClient.executeQueryAndGetResult(sql);
+ }
+
+ public ResultSet executeJsonQueryAndGetResult(final String json) throws ServiceException, IOException {
+ return queryClient.executeJsonQueryAndGetResult(json);
+ }
+
+ public QueryStatus getQueryStatus(QueryId queryId) throws ServiceException {
+ return queryClient.getQueryStatus(queryId);
+ }
+
+ public ResultSet getQueryResult(QueryId queryId) throws ServiceException, IOException {
+ return queryClient.getQueryResult(queryId);
+ }
+
+ public ResultSet createNullResultSet(QueryId queryId) throws IOException {
+ return new TajoResultSet(this, queryId);
+ }
+
+ public GetQueryResultResponse getResultResponse(QueryId queryId) throws ServiceException {
+ return queryClient.getResultResponse(queryId);
+ }
+
+ public TajoMemoryResultSet fetchNextQueryResult(final QueryId queryId, final int fetchRowNum) throws ServiceException {
+ return queryClient.fetchNextQueryResult(queryId, fetchRowNum);
+ }
+
+ public boolean updateQuery(final String sql) throws ServiceException {
+ return queryClient.updateQuery(sql);
+ }
+
+ public boolean updateQueryWithJson(final String json) throws ServiceException {
+ return queryClient.updateQueryWithJson(json);
+ }
+
+ public QueryStatus killQuery(final QueryId queryId) throws ServiceException, IOException {
+ return queryClient.killQuery(queryId);
+ }
+
+ public List<BriefQueryInfo> getRunningQueryList() throws ServiceException {
+ return queryClient.getRunningQueryList();
+ }
+
+ public List<BriefQueryInfo> getFinishedQueryList() throws ServiceException {
+ return queryClient.getFinishedQueryList();
+ }
+
+ public List<WorkerResourceInfo> getClusterInfo() throws ServiceException {
+ return queryClient.getClusterInfo();
+ }
+
+ /*------------------------------------------------------------------------*/
+ // CatalogClient wrappers
+ /*------------------------------------------------------------------------*/
+
+ public boolean createDatabase(final String databaseName) throws ServiceException {
+ return catalogClient.createDatabase(databaseName);
+ }
+
+ public boolean existDatabase(final String databaseName) throws ServiceException {
+ return catalogClient.existDatabase(databaseName);
+ }
+
+ public boolean dropDatabase(final String databaseName) throws ServiceException {
+ return catalogClient.dropDatabase(databaseName);
+ }
+
+ public List<String> getAllDatabaseNames() throws ServiceException {
+ return catalogClient.getAllDatabaseNames();
+ }
+
+ public boolean existTable(final String tableName) throws ServiceException {
+ return catalogClient.existTable(tableName);
+ }
+
+ public TableDesc createExternalTable(final String tableName, final Schema schema, final Path path,
+ final TableMeta meta) throws SQLException, ServiceException {
+ return catalogClient.createExternalTable(tableName, schema, path, meta);
+ }
+
+ public TableDesc createExternalTable(final String tableName, final Schema schema, final Path path,
+ final TableMeta meta, final PartitionMethodDesc partitionMethodDesc)
+ throws SQLException, ServiceException {
+ return catalogClient.createExternalTable(tableName, schema, path, meta, partitionMethodDesc);
+ }
+
+ public boolean dropTable(final String tableName) throws ServiceException {
+ return dropTable(tableName, false);
+ }
+
+ public boolean dropTable(final String tableName, final boolean purge) throws ServiceException {
+ return catalogClient.dropTable(tableName, purge);
+ }
+
+ public List<String> getTableList(@Nullable final String databaseName) throws ServiceException {
+ return catalogClient.getTableList(databaseName);
+ }
+
+ public TableDesc getTableDesc(final String tableName) throws ServiceException {
+ return catalogClient.getTableDesc(tableName);
+ }
+
+ public List<CatalogProtos.FunctionDescProto> getFunctions(final String functionName) throws ServiceException {
+ return catalogClient.getFunctions(functionName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/TajoClientUtil.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/TajoClientUtil.java b/tajo-client/src/main/java/org/apache/tajo/client/TajoClientUtil.java
new file mode 100644
index 0000000..7aed335
--- /dev/null
+++ b/tajo-client/src/main/java/org/apache/tajo/client/TajoClientUtil.java
@@ -0,0 +1,94 @@
+/**
+ * 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.client;
+
+import org.apache.tajo.QueryId;
+import org.apache.tajo.SessionVars;
+import org.apache.tajo.TajoProtos;
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableDesc;
+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.rpc.protocolrecords.PrimitiveProtos;
+
+import java.io.IOException;
+import java.sql.ResultSet;
+
+public class TajoClientUtil {
+
+ /* query submit */
+ public static boolean isQueryWaitingForSchedule(TajoProtos.QueryState state) {
+ return state == TajoProtos.QueryState.QUERY_NOT_ASSIGNED ||
+ state == TajoProtos.QueryState.QUERY_MASTER_INIT ||
+ state == TajoProtos.QueryState.QUERY_MASTER_LAUNCHED;
+ }
+
+ /* query submitted. but is not running */
+ public static boolean isQueryInited(TajoProtos.QueryState state) {
+ return state == TajoProtos.QueryState.QUERY_NEW || state == TajoProtos.QueryState.QUERY_INIT;
+ }
+
+ /* query started. but is not complete */
+ public static boolean isQueryRunning(TajoProtos.QueryState state) {
+ return isQueryInited(state) || state == TajoProtos.QueryState.QUERY_RUNNING;
+ }
+
+ /* query complete */
+ public static boolean isQueryComplete(TajoProtos.QueryState state) {
+ return !isQueryWaitingForSchedule(state) && !isQueryRunning(state);
+ }
+
+ public static ResultSet createResultSet(TajoConf conf, TajoClient client, QueryId queryId,
+ ClientProtos.GetQueryResultResponse response)
+ throws IOException {
+ TableDesc desc = CatalogUtil.newTableDesc(response.getTableDesc());
+ conf.setVar(TajoConf.ConfVars.USERNAME, response.getTajoUserName());
+ return new TajoResultSet(client, queryId, conf, desc);
+ }
+
+ public static ResultSet createResultSet(TajoConf conf, QueryClient client, ClientProtos.SubmitQueryResponse response)
+ throws IOException {
+ if (response.hasTableDesc()) {
+ // non-forward query
+ // select * from table1 [limit 10]
+ int fetchRowNum = conf.getIntVar(TajoConf.ConfVars.$RESULT_SET_FETCH_ROWNUM);
+ if (response.hasSessionVariables()) {
+ for (PrimitiveProtos.KeyValueProto eachKeyValue: response.getSessionVariables().getKeyvalList()) {
+ if (eachKeyValue.getKey().equals(SessionVars.FETCH_ROWNUM.keyname())) {
+ fetchRowNum = Integer.parseInt(eachKeyValue.getValue());
+ }
+ }
+ }
+ TableDesc tableDesc = new TableDesc(response.getTableDesc());
+ return new FetchResultSet(client, tableDesc.getLogicalSchema(), new QueryId(response.getQueryId()), fetchRowNum);
+ } else {
+ // simple eval query
+ // select substr('abc', 1, 2)
+ ClientProtos.SerializedResultSet serializedResultSet = response.getResultSet();
+ return new TajoMemoryResultSet(
+ new Schema(serializedResultSet.getSchema()),
+ serializedResultSet.getSerializedTuplesList(),
+ response.getMaxRowNum());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/TajoDump.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/TajoDump.java b/tajo-client/src/main/java/org/apache/tajo/client/TajoDump.java
index 7628d9d..540f54b 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/TajoDump.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/TajoDump.java
@@ -104,9 +104,9 @@ public class TajoDump {
System.exit(-1);
} else if (hostName != null && port != null) {
conf.setVar(TajoConf.ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS, hostName+":"+port);
- client = new TajoClient(conf);
+ client = new TajoClientImpl(conf);
} else {
- client = new TajoClient(conf);
+ client = new TajoClientImpl(conf);
}
PrintWriter writer = new PrintWriter(System.out);
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/TajoGetConf.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/TajoGetConf.java b/tajo-client/src/main/java/org/apache/tajo/client/TajoGetConf.java
index 2377427..88ab491 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/TajoGetConf.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/TajoGetConf.java
@@ -112,9 +112,9 @@ public class TajoGetConf {
return;
} else if (hostName != null && port != null) {
tajoConf.setVar(TajoConf.ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS, hostName + ":" + port);
- tajoClient = new TajoClient(tajoConf);
+ tajoClient = new TajoClientImpl(tajoConf);
} else if (hostName == null && port == null) {
- tajoClient = new TajoClient(tajoConf);
+ tajoClient = new TajoClientImpl(tajoConf);
}
processConfKey(writer, param);
@@ -123,13 +123,13 @@ public class TajoGetConf {
private void processConfKey(Writer writer, String param) throws ParseException, IOException,
ServiceException, SQLException {
- String value = tajoClient.getConf().getTrimmed(param);
+ String value = tajoConf.getTrimmed(param);
// If there is no value in the configuration file, we need to find all ConfVars.
if (value == null) {
for(TajoConf.ConfVars vars : TajoConf.ConfVars.values()) {
if (vars.varname.equalsIgnoreCase(param)) {
- value = tajoClient.getConf().getVar(vars);
+ value = tajoConf.getVar(vars);
break;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/TajoHAAdmin.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/TajoHAAdmin.java b/tajo-client/src/main/java/org/apache/tajo/client/TajoHAAdmin.java
index 11cb4ed..5d5cf71 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/TajoHAAdmin.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/TajoHAAdmin.java
@@ -126,9 +126,9 @@ public class TajoHAAdmin {
return;
} else if (hostName != null && port != null) {
tajoConf.setVar(TajoConf.ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS, hostName + ":" + port);
- tajoClient = new TajoClient(tajoConf);
+ tajoClient = new TajoClientImpl(tajoConf);
} else if (hostName == null && port == null) {
- tajoClient = new TajoClient(tajoConf);
+ tajoClient = new TajoClientImpl(tajoConf);
}
if (!tajoConf.getBoolVar(TajoConf.ConfVars.TAJO_MASTER_HA_ENABLE)) {
@@ -159,7 +159,7 @@ public class TajoHAAdmin {
private void getState(Writer writer, String param) throws ParseException, IOException,
ServiceException {
tajoClient = TajoHAClientUtil.getTajoClient(tajoConf, tajoClient);
- int retValue = HAServiceUtil.getState(param, tajoClient.getConf());
+ int retValue = HAServiceUtil.getState(param, tajoConf);
switch (retValue) {
case 1:
@@ -179,7 +179,7 @@ public class TajoHAAdmin {
private void formatHA(Writer writer) throws ParseException, IOException,
ServiceException {
- int retValue = HAServiceUtil.formatHA(tajoClient.getConf());
+ int retValue = HAServiceUtil.formatHA(tajoConf);
switch (retValue) {
case 1:
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/TajoHAClientUtil.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/TajoHAClientUtil.java b/tajo-client/src/main/java/org/apache/tajo/client/TajoHAClientUtil.java
index f22d5ba..b93590c 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/TajoHAClientUtil.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/TajoHAClientUtil.java
@@ -71,7 +71,7 @@ public class TajoHAClientUtil {
conf.setVar(TajoConf.ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS,
HAServiceUtil.getMasterClientName(conf));
client.close();
- tajoClient = new TajoClient(conf, baseDatabase);
+ tajoClient = new TajoClientImpl(conf, baseDatabase);
if (context != null && context.getCurrentDatabase() != null) {
tajoClient.selectDatabase(context.getCurrentDatabase());
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/jdbc/FetchResultSet.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/jdbc/FetchResultSet.java b/tajo-client/src/main/java/org/apache/tajo/jdbc/FetchResultSet.java
index 7ebce91..78674b1 100644
--- a/tajo-client/src/main/java/org/apache/tajo/jdbc/FetchResultSet.java
+++ b/tajo-client/src/main/java/org/apache/tajo/jdbc/FetchResultSet.java
@@ -20,6 +20,7 @@ package org.apache.tajo.jdbc;
import org.apache.tajo.QueryId;
import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.client.QueryClient;
import org.apache.tajo.client.TajoClient;
import org.apache.tajo.storage.Tuple;
@@ -27,13 +28,13 @@ import java.io.IOException;
import java.sql.SQLException;
public class FetchResultSet extends TajoResultSetBase {
- private TajoClient tajoClient;
+ private QueryClient tajoClient;
private QueryId queryId;
private int fetchRowNum;
private TajoMemoryResultSet currentResultSet;
private boolean finished = false;
- public FetchResultSet(TajoClient tajoClient, Schema schema, QueryId queryId, int fetchRowNum) {
+ public FetchResultSet(QueryClient tajoClient, Schema schema, QueryId queryId, int fetchRowNum) {
this.tajoClient = tajoClient;
this.queryId = queryId;
this.fetchRowNum = fetchRowNum;
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java b/tajo-client/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java
index 65954f1..d78b04f 100644
--- a/tajo-client/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java
+++ b/tajo-client/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.fs.PathFilter;
import org.apache.tajo.QueryId;
import org.apache.tajo.QueryIdFactory;
import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.client.QueryClient;
import org.apache.tajo.client.TajoClient;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.storage.FileScanner;
@@ -47,20 +48,20 @@ public class TajoResultSet extends TajoResultSetBase {
private FileSystem fs;
private Scanner scanner;
- private TajoClient tajoClient;
+ private QueryClient tajoClient;
private TajoConf conf;
private TableDesc desc;
private Long maxRowNum = null;
private QueryId queryId;
private AtomicBoolean closed = new AtomicBoolean(false);
- public TajoResultSet(TajoClient tajoClient, QueryId queryId) {
+ public TajoResultSet(QueryClient tajoClient, QueryId queryId) {
this.tajoClient = tajoClient;
this.queryId = queryId;
init();
}
- public TajoResultSet(TajoClient tajoClient, QueryId queryId, TajoConf conf, TableDesc table) throws IOException {
+ public TajoResultSet(QueryClient tajoClient, QueryId queryId, TajoConf conf, TableDesc table) throws IOException {
this.tajoClient = tajoClient;
this.queryId = queryId;
this.conf = conf;
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-core/src/main/java/org/apache/tajo/benchmark/BenchmarkSet.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/benchmark/BenchmarkSet.java b/tajo-core/src/main/java/org/apache/tajo/benchmark/BenchmarkSet.java
index 91dcea1..b1b6450 100644
--- a/tajo-core/src/main/java/org/apache/tajo/benchmark/BenchmarkSet.java
+++ b/tajo-core/src/main/java/org/apache/tajo/benchmark/BenchmarkSet.java
@@ -24,6 +24,7 @@ import org.apache.tajo.catalog.CatalogConstants;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.store.MemStore;
import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.TajoClientImpl;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.conf.TajoConf.ConfVars;
import org.apache.tajo.util.FileUtil;
@@ -43,11 +44,11 @@ public abstract class BenchmarkSet {
public void init(TajoConf conf, String dataDir) throws IOException {
this.dataDir = dataDir;
if (System.getProperty(ConfVars.WORKER_PEER_RPC_ADDRESS.varname) != null) {
- tajo = new TajoClient(NetUtils.createSocketAddr(
+ tajo = new TajoClientImpl(NetUtils.createSocketAddr(
System.getProperty(ConfVars.WORKER_PEER_RPC_ADDRESS.varname)));
} else {
conf.set(CatalogConstants.STORE_CLASS, MemStore.class.getCanonicalName());
- tajo = new TajoClient(conf);
+ tajo = new TajoClientImpl(conf);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java b/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
index cd3be98..f7c7b11 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
@@ -41,7 +41,7 @@ import org.apache.tajo.catalog.exception.*;
import org.apache.tajo.catalog.partition.PartitionMethodDesc;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.statistics.TableStats;
-import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.QueryClient;
import org.apache.tajo.common.TajoDataTypes;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.datum.DatumFactory;
@@ -761,7 +761,7 @@ public class GlobalEngine extends AbstractService {
stats.setNumBytes(totalSize);
if (isExternal) { // if it is an external table, there is no way to know the exact row number without processing.
- stats.setNumRows(TajoClient.UNKNOWN_ROW_NUMBER);
+ stats.setNumRows(QueryClient.UNKNOWN_ROW_NUMBER);
}
TableDesc desc = new TableDesc(CatalogUtil.buildFQName(databaseName, simpleTableName),
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-core/src/main/java/org/apache/tajo/webapp/QueryExecutorServlet.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/webapp/QueryExecutorServlet.java b/tajo-core/src/main/java/org/apache/tajo/webapp/QueryExecutorServlet.java
index 7ea2e48..7790ac6 100644
--- a/tajo-core/src/main/java/org/apache/tajo/webapp/QueryExecutorServlet.java
+++ b/tajo-core/src/main/java/org/apache/tajo/webapp/QueryExecutorServlet.java
@@ -9,13 +9,10 @@ import org.apache.tajo.QueryIdFactory;
import org.apache.tajo.TajoProtos;
import org.apache.tajo.catalog.CatalogUtil;
import org.apache.tajo.catalog.TableDesc;
-import org.apache.tajo.client.QueryStatus;
-import org.apache.tajo.client.TajoClient;
-import org.apache.tajo.client.TajoHAClientUtil;
+import org.apache.tajo.client.*;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.ipc.ClientProtos;
import org.apache.tajo.jdbc.TajoResultSet;
-import org.apache.tajo.util.HAServiceUtil;
import org.apache.tajo.util.JSPUtil;
import org.apache.tajo.util.TajoIdUtils;
import org.codehaus.jackson.map.DeserializationConfig;
@@ -67,18 +64,21 @@ public class QueryExecutorServlet extends HttpServlet {
//queryRunnerId -> QueryRunner
private final Map<String, QueryRunner> queryRunners = new HashMap<String, QueryRunner>();
+ private TajoConf tajoConf;
private TajoClient tajoClient;
private ExecutorService queryRunnerExecutor = Executors.newFixedThreadPool(5);
private QueryRunnerCleaner queryRunnerCleaner;
+
@Override
public void init(ServletConfig config) throws ServletException {
om.getDeserializationConfig().disable(
DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES);
try {
- tajoClient = new TajoClient(new TajoConf());
+ tajoConf = new TajoConf();
+ tajoClient = new TajoClientImpl(tajoConf);
queryRunnerCleaner = new QueryRunnerCleaner();
queryRunnerCleaner.start();
@@ -273,8 +273,7 @@ public class QueryExecutorServlet extends HttpServlet {
public void run() {
startTime = System.currentTimeMillis();
try {
- TajoConf conf = tajoClient.getConf();
- tajoClient = TajoHAClientUtil.getTajoClient(conf, tajoClient);
+ tajoClient = TajoHAClientUtil.getTajoClient(tajoConf, tajoClient);
response = tajoClient.executeQuery(query);
@@ -319,7 +318,7 @@ public class QueryExecutorServlet extends HttpServlet {
// non-forwarded INSERT INTO query does not have any query id.
// In this case, it just returns succeeded query information without printing the query results.
} else {
- res = TajoClient.createResultSet(tajoClient, response);
+ res = TajoClientUtil.createResultSet(tajoConf, tajoClient, response);
MakeResultText(res, desc);
}
progress.set(100);
@@ -399,8 +398,8 @@ public class QueryExecutorServlet extends HttpServlet {
try {
ClientProtos.GetQueryResultResponse response = tajoClient.getResultResponse(tajoQueryId);
TableDesc desc = CatalogUtil.newTableDesc(response.getTableDesc());
- tajoClient.getConf().setVar(TajoConf.ConfVars.USERNAME, response.getTajoUserName());
- res = new TajoResultSet(tajoClient, queryId, tajoClient.getConf(), desc);
+ tajoConf.setVar(TajoConf.ConfVars.USERNAME, response.getTajoUserName());
+ res = new TajoResultSet(tajoClient, queryId, tajoConf, desc);
MakeResultText(res, desc);
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-core/src/test/java/org/apache/tajo/LocalTajoTestingUtility.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/LocalTajoTestingUtility.java b/tajo-core/src/test/java/org/apache/tajo/LocalTajoTestingUtility.java
index 805fe06..c6cac32 100644
--- a/tajo-core/src/test/java/org/apache/tajo/LocalTajoTestingUtility.java
+++ b/tajo-core/src/test/java/org/apache/tajo/LocalTajoTestingUtility.java
@@ -31,6 +31,7 @@ import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.TajoClientImpl;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.engine.planner.global.MasterPlan;
import org.apache.tajo.engine.query.QueryContext;
@@ -102,7 +103,7 @@ public class LocalTajoTestingUtility {
util = new TajoTestingCluster();
util.startMiniCluster(1);
conf = util.getConfiguration();
- client = new TajoClient(conf);
+ client = new TajoClientImpl(conf);
FileSystem fs = util.getDefaultFileSystem();
Path rootDir = util.getMaster().getStorageManager().getWarehouseDir();
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-core/src/test/java/org/apache/tajo/QueryTestCaseBase.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/QueryTestCaseBase.java b/tajo-core/src/test/java/org/apache/tajo/QueryTestCaseBase.java
index a272b15..becb73e 100644
--- a/tajo-core/src/test/java/org/apache/tajo/QueryTestCaseBase.java
+++ b/tajo-core/src/test/java/org/apache/tajo/QueryTestCaseBase.java
@@ -32,6 +32,7 @@ import org.apache.tajo.catalog.TableDesc;
import org.apache.tajo.cli.ParsedResult;
import org.apache.tajo.cli.SimpleParser;
import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.TajoClientImpl;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.engine.parser.SQLAnalyzer;
import org.apache.tajo.engine.query.QueryContext;
@@ -178,7 +179,7 @@ public class QueryTestCaseBase {
@BeforeClass
public static void setUpClass() throws IOException {
conf = testBase.getTestingCluster().getConfiguration();
- client = new TajoClient(conf);
+ client = new TajoClientImpl(conf);
}
@AfterClass
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-core/src/test/java/org/apache/tajo/TajoTestingCluster.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/TajoTestingCluster.java b/tajo-core/src/test/java/org/apache/tajo/TajoTestingCluster.java
index aec11f6..452a17e 100644
--- a/tajo-core/src/test/java/org/apache/tajo/TajoTestingCluster.java
+++ b/tajo-core/src/test/java/org/apache/tajo/TajoTestingCluster.java
@@ -35,6 +35,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.proto.CatalogProtos;
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.conf.TajoConf.ConfVars;
import org.apache.tajo.master.TajoMaster;
@@ -583,7 +585,7 @@ public class TajoTestingCluster {
Thread.sleep(1000);
}
TajoConf conf = util.getConfiguration();
- return new TajoClient(conf);
+ return new TajoClientImpl(conf);
}
public static ResultSet run(String[] names,
@@ -620,7 +622,7 @@ public class TajoTestingCluster {
Thread.sleep(1000);
}
TajoConf conf = util.getConfiguration();
- TajoClient client = new TajoClient(conf);
+ TajoClient client = new TajoClientImpl(conf);
try {
return run(names, schemas, tableOption, tables, query, client);
@@ -645,7 +647,7 @@ public class TajoTestingCluster {
Thread.sleep(1000);
}
TajoConf conf = util.getConfiguration();
- TajoClient client = new TajoClient(conf);
+ TajoClient client = new TajoClientImpl(conf);
try {
FileSystem fs = util.getDefaultFileSystem();
Path rootDir = util.getMaster().
@@ -721,7 +723,7 @@ public class TajoTestingCluster {
QueryMasterTask qmt = null;
int i = 0;
- while (qmt == null || TajoClient.isInPreNewState(qmt.getState())) {
+ while (qmt == null || TajoClientUtil.isQueryWaitingForSchedule(qmt.getState())) {
try {
Thread.sleep(delay);
if(qmt == null){
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java b/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java
index 4ede88e..719a775 100644
--- a/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java
+++ b/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java
@@ -62,7 +62,7 @@ public class TestTajoClient {
public static void setUp() throws Exception {
cluster = TpchTestBase.getInstance().getTestingCluster();
conf = cluster.getConfiguration();
- client = new TajoClient(conf);
+ client = new TajoClientImpl(conf);
testDir = CommonTestingUtil.getTestDir();
}
@@ -658,7 +658,7 @@ public class TestTajoClient {
QueryStatus queryStatus = client.getQueryStatus(queryId);
assertNotNull(queryStatus);
- assertTrue(TajoClient.isInCompleteState(queryStatus.getState()));
+ assertTrue(TajoClientUtil.isQueryComplete(queryStatus.getState()));
TajoResultSet resultSet = (TajoResultSet) client.getQueryResult(queryId);
assertNotNull(resultSet);
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-core/src/test/java/org/apache/tajo/jdbc/TestTajoJdbc.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/jdbc/TestTajoJdbc.java b/tajo-core/src/test/java/org/apache/tajo/jdbc/TestTajoJdbc.java
index e477939..08535ef 100644
--- a/tajo-core/src/test/java/org/apache/tajo/jdbc/TestTajoJdbc.java
+++ b/tajo-core/src/test/java/org/apache/tajo/jdbc/TestTajoJdbc.java
@@ -25,7 +25,7 @@ import org.apache.tajo.TajoConstants;
import org.apache.tajo.catalog.CatalogUtil;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.TableDesc;
-import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.QueryClient;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -419,7 +419,7 @@ public class TestTajoJdbc extends QueryTestCaseBase {
assertNotNull(rsmd);
assertEquals(0, rsmd.getColumnCount());
- TajoClient connTajoClient = ((TajoConnection)stmt.getConnection()).getTajoClient();
+ QueryClient connTajoClient = ((JdbcConnection)stmt.getConnection()).getQueryClient();
Map<String, String> variables = connTajoClient.getAllSessionVariables();
String value = variables.get("JOIN_TASK_INPUT_SIZE");
assertNotNull(value);
@@ -461,7 +461,7 @@ public class TestTajoJdbc extends QueryTestCaseBase {
assertNotNull(rsmd);
assertEquals(0, rsmd.getColumnCount());
- TajoClient connTajoClient = ((TajoConnection)stmt.getConnection()).getTajoClient();
+ QueryClient connTajoClient = ((JdbcConnection)stmt.getConnection()).getQueryClient();
Map<String, String> variables = connTajoClient.getAllSessionVariables();
String value = variables.get("JOIN_TASK_INPUT_SIZE");
assertNotNull(value);
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-core/src/test/java/org/apache/tajo/master/ha/TestHAServiceHDFSImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/master/ha/TestHAServiceHDFSImpl.java b/tajo-core/src/test/java/org/apache/tajo/master/ha/TestHAServiceHDFSImpl.java
index 86d18eb..249afae 100644
--- a/tajo-core/src/test/java/org/apache/tajo/master/ha/TestHAServiceHDFSImpl.java
+++ b/tajo-core/src/test/java/org/apache/tajo/master/ha/TestHAServiceHDFSImpl.java
@@ -26,6 +26,7 @@ import org.apache.hadoop.net.NetUtils;
import org.apache.tajo.TajoConstants;
import org.apache.tajo.TajoTestingCluster;
import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.TajoClientImpl;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.master.TajoMaster;
import org.junit.Test;
@@ -55,7 +56,7 @@ public class TestHAServiceHDFSImpl {
cluster.startMiniCluster(1);
conf = cluster.getConfiguration();
- client = new TajoClient(conf);
+ client = new TajoClientImpl(conf);
FileSystem fs = cluster.getDefaultFileSystem();
startBackupMasters();
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-core/src/test/java/org/apache/tajo/master/querymaster/TestQueryProgress.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/master/querymaster/TestQueryProgress.java b/tajo-core/src/test/java/org/apache/tajo/master/querymaster/TestQueryProgress.java
index 0f925bb..4a6ca00 100644
--- a/tajo-core/src/test/java/org/apache/tajo/master/querymaster/TestQueryProgress.java
+++ b/tajo-core/src/test/java/org/apache/tajo/master/querymaster/TestQueryProgress.java
@@ -21,6 +21,8 @@ package org.apache.tajo.master.querymaster;
import org.apache.tajo.*;
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;
@@ -41,7 +43,7 @@ public class TestQueryProgress {
public static void setUp() throws Exception {
cluster = TpchTestBase.getInstance().getTestingCluster();
conf = cluster.getConfiguration();
- client = new TajoClient(conf);
+ client = new TajoClientImpl(conf);
}
@AfterClass
@@ -67,7 +69,7 @@ public class TestQueryProgress {
prevProgress = progress;
assertTrue(progress <= 1.0f);
- if (TajoClient.isInCompleteState(status.getState())) break;
+ if (TajoClientUtil.isQueryComplete(status.getState())) break;
}
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-core/src/test/java/org/apache/tajo/scheduler/TestFifoScheduler.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/scheduler/TestFifoScheduler.java b/tajo-core/src/test/java/org/apache/tajo/scheduler/TestFifoScheduler.java
index 069ee27..18764c2 100644
--- a/tajo-core/src/test/java/org/apache/tajo/scheduler/TestFifoScheduler.java
+++ b/tajo-core/src/test/java/org/apache/tajo/scheduler/TestFifoScheduler.java
@@ -20,6 +20,8 @@ package org.apache.tajo.scheduler;
import org.apache.tajo.*;
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.junit.AfterClass;
@@ -41,7 +43,7 @@ public class TestFifoScheduler {
public static void setUp() throws Exception {
cluster = TpchTestBase.getInstance().getTestingCluster();
conf = cluster.getConfiguration();
- client = new TajoClient(conf);
+ client = new TajoClientImpl(conf);
}
@AfterClass
@@ -75,7 +77,7 @@ public class TestFifoScheduler {
cluster.waitForQueryRunning(queryId);
assertEquals(TajoProtos.QueryState.QUERY_SUCCEEDED, client.getQueryStatus(queryId2).getState());
- ResultSet resSet = TajoClient.createResultSet(client, res2);
+ ResultSet resSet = TajoClientUtil.createResultSet(conf, client, res2);
assertNotNull(resSet);
client.killQuery(queryId); //cleanup
@@ -95,7 +97,7 @@ public class TestFifoScheduler {
cluster.waitForQueryRunning(queryId);
- assertTrue(TajoClient.isInRunningState(client.getQueryStatus(queryId).getState()));
+ assertTrue(TajoClientUtil.isQueryRunning(client.getQueryStatus(queryId).getState()));
assertEquals(TajoProtos.QueryState.QUERY_MASTER_INIT, client.getQueryStatus(queryId2).getState());
assertEquals(TajoProtos.QueryState.QUERY_MASTER_INIT, client.getQueryStatus(queryId3).getState());
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-core/src/test/java/org/apache/tajo/worker/TestHistory.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/worker/TestHistory.java b/tajo-core/src/test/java/org/apache/tajo/worker/TestHistory.java
index 3a85c14..c68d3a4 100644
--- a/tajo-core/src/test/java/org/apache/tajo/worker/TestHistory.java
+++ b/tajo-core/src/test/java/org/apache/tajo/worker/TestHistory.java
@@ -25,6 +25,7 @@ import org.apache.tajo.TajoProtos;
import org.apache.tajo.TajoTestingCluster;
import org.apache.tajo.TpchTestBase;
import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.TajoClientImpl;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.master.TajoMaster;
import org.apache.tajo.master.querymaster.QueryInProgress;
@@ -49,7 +50,7 @@ public class TestHistory {
cluster = TpchTestBase.getInstance().getTestingCluster();
master = cluster.getMaster();
conf = cluster.getConfiguration();
- client = new TajoClient(conf);
+ client = new TajoClientImpl(conf);
}
@After
[3/3] git commit: TAJO-1140: Separate TajoClient into fine grained
parts.
Posted by hy...@apache.org.
TAJO-1140: Separate TajoClient into fine grained parts.
Closes #213
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/f80beaf6
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/f80beaf6
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/f80beaf6
Branch: refs/heads/master
Commit: f80beaf64d33850ee79e8ae32b33a852f56712f0
Parents: 8741e68
Author: Hyunsik Choi <hy...@apache.org>
Authored: Mon Oct 27 22:16:30 2014 -0700
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Mon Oct 27 22:16:30 2014 -0700
----------------------------------------------------------------------
CHANGES | 2 +
.../tajo/cli/DefaultTajoCliOutputFormatter.java | 6 +-
.../org/apache/tajo/cli/DescTableCommand.java | 4 +-
.../tajo/cli/InvalidClientSessionException.java | 27 -
.../main/java/org/apache/tajo/cli/TajoCli.java | 20 +-
.../org/apache/tajo/cli/TajoShellCommand.java | 2 +-
.../apache/tajo/client/CatalogAdminClient.java | 137 +++
.../tajo/client/CatalogAdminClientImpl.java | 266 +++++
.../client/InvalidClientSessionException.java | 27 +
.../org/apache/tajo/client/QueryClient.java | 116 ++
.../org/apache/tajo/client/QueryClientImpl.java | 622 +++++++++++
.../apache/tajo/client/SessionConnection.java | 326 ++++++
.../java/org/apache/tajo/client/TajoAdmin.java | 8 +-
.../java/org/apache/tajo/client/TajoClient.java | 1023 +-----------------
.../org/apache/tajo/client/TajoClientImpl.java | 215 ++++
.../org/apache/tajo/client/TajoClientUtil.java | 94 ++
.../java/org/apache/tajo/client/TajoDump.java | 4 +-
.../org/apache/tajo/client/TajoGetConf.java | 8 +-
.../org/apache/tajo/client/TajoHAAdmin.java | 8 +-
.../apache/tajo/client/TajoHAClientUtil.java | 2 +-
.../org/apache/tajo/jdbc/FetchResultSet.java | 5 +-
.../org/apache/tajo/jdbc/TajoResultSet.java | 7 +-
.../org/apache/tajo/benchmark/BenchmarkSet.java | 5 +-
.../org/apache/tajo/master/GlobalEngine.java | 4 +-
.../tajo/webapp/QueryExecutorServlet.java | 19 +-
.../apache/tajo/LocalTajoTestingUtility.java | 3 +-
.../java/org/apache/tajo/QueryTestCaseBase.java | 3 +-
.../org/apache/tajo/TajoTestingCluster.java | 10 +-
.../org/apache/tajo/client/TestTajoClient.java | 4 +-
.../java/org/apache/tajo/jdbc/TestTajoJdbc.java | 6 +-
.../tajo/master/ha/TestHAServiceHDFSImpl.java | 3 +-
.../master/querymaster/TestQueryProgress.java | 6 +-
.../tajo/scheduler/TestFifoScheduler.java | 8 +-
.../org/apache/tajo/worker/TestHistory.java | 3 +-
.../org/apache/tajo/jdbc/JdbcConnection.java | 444 ++++++++
.../org/apache/tajo/jdbc/TajoConnection.java | 437 --------
.../apache/tajo/jdbc/TajoDatabaseMetaData.java | 27 +-
.../java/org/apache/tajo/jdbc/TajoDriver.java | 2 +-
.../apache/tajo/jdbc/TajoPreparedStatement.java | 4 +-
.../org/apache/tajo/jdbc/TajoStatement.java | 5 +-
40 files changed, 2346 insertions(+), 1576 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index d239a19..ba2646d 100644
--- a/CHANGES
+++ b/CHANGES
@@ -11,6 +11,8 @@ Release 0.9.1 - unreleased
IMPROVEMENT
+ TAJO-1140: Separate TajoClient into fine grained parts. (hyunsik)
+
TAJO-1132: More detailed version info in tsql. (hyunsik)
TAJO-1125: Separate logical plan and optimizer into a maven module.
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/cli/DefaultTajoCliOutputFormatter.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/cli/DefaultTajoCliOutputFormatter.java b/tajo-client/src/main/java/org/apache/tajo/cli/DefaultTajoCliOutputFormatter.java
index 5797778..c583aa2 100644
--- a/tajo-client/src/main/java/org/apache/tajo/cli/DefaultTajoCliOutputFormatter.java
+++ b/tajo-client/src/main/java/org/apache/tajo/cli/DefaultTajoCliOutputFormatter.java
@@ -23,8 +23,8 @@ import org.apache.tajo.QueryId;
import org.apache.tajo.SessionVars;
import org.apache.tajo.catalog.TableDesc;
import org.apache.tajo.catalog.statistics.TableStats;
+import org.apache.tajo.client.QueryClient;
import org.apache.tajo.client.QueryStatus;
-import org.apache.tajo.client.TajoClient;
import org.apache.tajo.util.FileUtil;
import java.io.InputStream;
@@ -56,10 +56,10 @@ public class DefaultTajoCliOutputFormatter implements TajoCliOutputFormatter {
TableStats stat = tableDesc.getStats();
String volume = stat == null ? (endOfTuple ? "0 B" : "unknown bytes") :
FileUtil.humanReadableByteCount(stat.getNumBytes(), false);
- long resultRows = stat == null ? TajoClient.UNKNOWN_ROW_NUMBER : stat.getNumRows();
+ long resultRows = stat == null ? QueryClient.UNKNOWN_ROW_NUMBER : stat.getNumRows();
String displayRowNum;
- if (resultRows == TajoClient.UNKNOWN_ROW_NUMBER) {
+ if (resultRows == QueryClient.UNKNOWN_ROW_NUMBER) {
if (endOfTuple) {
displayRowNum = totalPrintedRows + " rows";
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/cli/DescTableCommand.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/cli/DescTableCommand.java b/tajo-client/src/main/java/org/apache/tajo/cli/DescTableCommand.java
index 8fab138..4b34858 100644
--- a/tajo-client/src/main/java/org/apache/tajo/cli/DescTableCommand.java
+++ b/tajo-client/src/main/java/org/apache/tajo/cli/DescTableCommand.java
@@ -23,7 +23,7 @@ import org.apache.commons.lang.StringEscapeUtils;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.TableDesc;
import org.apache.tajo.catalog.partition.PartitionMethodDesc;
-import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.client.QueryClient;
import org.apache.tajo.util.FileUtil;
import org.apache.tajo.util.TUtil;
@@ -82,7 +82,7 @@ public class DescTableCommand extends TajoShellCommand {
if (desc.getStats() != null) {
long row = desc.getStats().getNumRows();
- String rowText = row == TajoClient.UNKNOWN_ROW_NUMBER ? "unknown" : row + "";
+ String rowText = row == QueryClient.UNKNOWN_ROW_NUMBER ? "unknown" : row + "";
sb.append("number of rows: ").append(rowText).append("\n");
sb.append("volume: ").append(
FileUtil.humanReadableByteCount(desc.getStats().getNumBytes(),
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/cli/InvalidClientSessionException.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/cli/InvalidClientSessionException.java b/tajo-client/src/main/java/org/apache/tajo/cli/InvalidClientSessionException.java
deleted file mode 100644
index 5c6c96e..0000000
--- a/tajo-client/src/main/java/org/apache/tajo/cli/InvalidClientSessionException.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/**
- * 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.cli;
-
-import com.google.protobuf.ServiceException;
-
-public class InvalidClientSessionException extends ServiceException {
- public InvalidClientSessionException(String message) {
- super(message);
- }
-}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/cli/TajoCli.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/cli/TajoCli.java b/tajo-client/src/main/java/org/apache/tajo/cli/TajoCli.java
index 05b919c..c732fd9 100644
--- a/tajo-client/src/main/java/org/apache/tajo/cli/TajoCli.java
+++ b/tajo-client/src/main/java/org/apache/tajo/cli/TajoCli.java
@@ -21,17 +21,13 @@ package org.apache.tajo.cli;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.protobuf.ServiceException;
-
import jline.UnsupportedTerminal;
import jline.console.ConsoleReader;
-
import org.apache.commons.cli.*;
import org.apache.tajo.*;
import org.apache.tajo.TajoProtos.QueryState;
import org.apache.tajo.catalog.TableDesc;
-import org.apache.tajo.client.QueryStatus;
-import org.apache.tajo.client.TajoClient;
-import org.apache.tajo.client.TajoHAClientUtil;
+import org.apache.tajo.client.*;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.conf.TajoConf.ConfVars;
import org.apache.tajo.ipc.ClientProtos;
@@ -233,9 +229,9 @@ public class TajoCli {
throw new RuntimeException("cannot find valid Tajo server address");
} else if (hostName != null && port != null) {
conf.setVar(ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS, hostName+":"+port);
- client = new TajoClient(conf, baseDatabase);
+ client = new TajoClientImpl(conf, baseDatabase);
} else if (hostName == null && port == null) {
- client = new TajoClient(conf, baseDatabase);
+ client = new TajoClientImpl(conf, baseDatabase);
}
try {
@@ -565,7 +561,7 @@ public class TajoCli {
if (response.getMaxRowNum() < 0 && queryId.equals(QueryIdFactory.NULL_QUERY_ID)) {
displayFormatter.printResult(sout, sin, desc, responseTime, res);
} else {
- res = TajoClient.createResultSet(client, response);
+ res = TajoClientUtil.createResultSet(conf, client, response);
displayFormatter.printResult(sout, sin, desc, responseTime, res);
}
} catch (Throwable t) {
@@ -597,17 +593,17 @@ public class TajoCli {
while (true) {
// TODO - configurable
status = client.getQueryStatus(queryId);
- if(TajoClient.isInPreNewState(status.getState())) {
+ if(TajoClientUtil.isQueryWaitingForSchedule(status.getState())) {
Thread.sleep(Math.min(20 * initRetries, 1000));
initRetries++;
continue;
}
- if (TajoClient.isInRunningState(status.getState()) || status.getState() == QueryState.QUERY_SUCCEEDED) {
+ if (TajoClientUtil.isQueryRunning(status.getState()) || status.getState() == QueryState.QUERY_SUCCEEDED) {
displayFormatter.printProgress(sout, status);
}
- if (TajoClient.isInCompleteState(status.getState()) && status.getState() != QueryState.QUERY_KILL_WAIT) {
+ if (TajoClientUtil.isQueryComplete(status.getState()) && status.getState() != QueryState.QUERY_KILL_WAIT) {
break;
} else {
Thread.sleep(Math.min(200 * progressRetries, 1000));
@@ -626,7 +622,7 @@ public class TajoCli {
float responseTime = ((float)(status.getFinishTime() - status.getSubmitTime()) / 1000.0f);
ClientProtos.GetQueryResultResponse response = client.getResultResponse(queryId);
if (status.hasResult()) {
- res = TajoClient.createResultSet(client, queryId, response);
+ res = TajoClientUtil.createResultSet(conf, client, queryId, response);
TableDesc desc = new TableDesc(response.getTableDesc());
displayFormatter.printResult(sout, sin, desc, responseTime, res);
} else {
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/cli/TajoShellCommand.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/cli/TajoShellCommand.java b/tajo-client/src/main/java/org/apache/tajo/cli/TajoShellCommand.java
index 39f5377..138aec4 100644
--- a/tajo-client/src/main/java/org/apache/tajo/cli/TajoShellCommand.java
+++ b/tajo-client/src/main/java/org/apache/tajo/cli/TajoShellCommand.java
@@ -40,7 +40,7 @@ public abstract class TajoShellCommand {
protected int maxColumn;
public TajoShellCommand(TajoCli.TajoCliContext context) {
- maxColumn = context.getTajoClient().getConf().getIntVar(TajoConf.ConfVars.$CLI_MAX_COLUMN);
+ maxColumn = context.getConf().getIntVar(TajoConf.ConfVars.$CLI_MAX_COLUMN);
this.context = context;
client = context.getTajoClient();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClient.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClient.java b/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClient.java
new file mode 100644
index 0000000..a36fc0e
--- /dev/null
+++ b/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClient.java
@@ -0,0 +1,137 @@
+/**
+ * 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.client;
+
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.annotation.Nullable;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.catalog.partition.PartitionMethodDesc;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+
+import java.io.Closeable;
+import java.sql.SQLException;
+import java.util.List;
+
+public interface CatalogAdminClient extends Closeable {
+ /**
+ * Create a database.
+ *
+ * @param databaseName The database name to be created. This name is case sensitive.
+ * @return True if created successfully.
+ * @throws com.google.protobuf.ServiceException
+ */
+ public boolean createDatabase(final String databaseName) throws ServiceException;
+ /**
+ * Does the database exist?
+ *
+ * @param databaseName The database name to be checked. This name is case sensitive.
+ * @return True if so.
+ * @throws ServiceException
+ */
+ public boolean existDatabase(final String databaseName) throws ServiceException;
+ /**
+ * Drop the database
+ *
+ * @param databaseName The database name to be dropped. This name is case sensitive.
+ * @return True if the database is dropped successfully.
+ * @throws ServiceException
+ */
+ public boolean dropDatabase(final String databaseName) throws ServiceException;
+
+ public List<String> getAllDatabaseNames() throws ServiceException;
+
+ /**
+ * Does the table exist?
+ *
+ * @param tableName The table name to be checked. This name is case sensitive.
+ * @return True if so.
+ */
+ public boolean existTable(final String tableName) throws ServiceException;
+
+ /**
+ * Create an external table.
+ *
+ * @param tableName The table name to be created. This name is case sensitive. This name can be qualified or not.
+ * If the table name is not qualified, the current database in the session will be used.
+ * @param schema The schema
+ * @param path The external table location
+ * @param meta Table meta
+ * @return the created table description.
+ * @throws java.sql.SQLException
+ * @throws ServiceException
+ */
+ public TableDesc createExternalTable(final String tableName, final Schema schema, final Path path,
+ final TableMeta meta) throws SQLException, ServiceException;
+
+ /**
+ * Create an external table.
+ *
+ * @param tableName The table name to be created. This name is case sensitive. This name can be qualified or not.
+ * If the table name is not qualified, the current database in the session will be used.
+ * @param schema The schema
+ * @param path The external table location
+ * @param meta Table meta
+ * @param partitionMethodDesc Table partition description
+ * @return the created table description.
+ * @throws SQLException
+ * @throws ServiceException
+ */
+ public TableDesc createExternalTable(final String tableName, final Schema schema, final Path path,
+ final TableMeta meta, final PartitionMethodDesc partitionMethodDesc)
+ throws SQLException, ServiceException;
+
+ /**
+ * Drop a table
+ *
+ * @param tableName The table name to be dropped. This name is case sensitive.
+ * @return True if the table is dropped successfully.
+ */
+ public boolean dropTable(final String tableName) throws ServiceException;
+
+ /**
+ * Drop a table.
+ *
+ * @param tableName The table name to be dropped. This name is case sensitive.
+ * @param purge If purge is true, this call will remove the entry in catalog as well as the table contents.
+ * @return True if the table is dropped successfully.
+ */
+ public boolean dropTable(final String tableName, final boolean purge) throws ServiceException;
+
+ /**
+ * Get a list of table names.
+ *
+ * @param databaseName The database name to show all tables. This name is case sensitive.
+ * If it is null, this method will show all tables
+ * in the current database of this session.
+ */
+ public List<String> getTableList(@Nullable final String databaseName) throws ServiceException;
+
+ /**
+ * Get a table description
+ *
+ * @param tableName The table name to get. This name is case sensitive.
+ * @return Table description
+ */
+ public TableDesc getTableDesc(final String tableName) throws ServiceException;
+
+ public List<CatalogProtos.FunctionDescProto> getFunctions(final String functionName) throws ServiceException;
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClientImpl.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClientImpl.java b/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClientImpl.java
new file mode 100644
index 0000000..496161d
--- /dev/null
+++ b/tajo-client/src/main/java/org/apache/tajo/client/CatalogAdminClientImpl.java
@@ -0,0 +1,266 @@
+/**
+ * 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.client;
+
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.annotation.Nullable;
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.catalog.partition.PartitionMethodDesc;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.ipc.ClientProtos;
+import org.apache.tajo.ipc.TajoMasterClientProtocol;
+import org.apache.tajo.jdbc.SQLStates;
+import org.apache.tajo.rpc.NettyClientBase;
+import org.apache.tajo.rpc.ServerCallable;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.List;
+
+import static org.apache.tajo.ipc.TajoMasterClientProtocol.TajoMasterClientProtocolService;
+import static org.apache.tajo.ipc.TajoMasterClientProtocol.TajoMasterClientProtocolService.BlockingInterface;
+
+public class CatalogAdminClientImpl implements CatalogAdminClient {
+ private final SessionConnection connection;
+
+ public CatalogAdminClientImpl(SessionConnection connection) {
+ this.connection = connection;
+ }
+
+ @Override
+ public boolean createDatabase(final String databaseName) throws ServiceException {
+ return new ServerCallable<Boolean>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public Boolean call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+ BlockingInterface tajoMaster = client.getStub();
+ return tajoMaster.createDatabase(null, connection.convertSessionedString(databaseName)).getValue();
+ }
+
+ }.withRetries();
+ }
+
+ @Override
+ public boolean existDatabase(final String databaseName) throws ServiceException {
+
+ return new ServerCallable<Boolean>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public Boolean call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+ BlockingInterface tajoMaster = client.getStub();
+ return tajoMaster.existDatabase(null, connection.convertSessionedString(databaseName)).getValue();
+ }
+
+ }.withRetries();
+ }
+
+ @Override
+ public boolean dropDatabase(final String databaseName) throws ServiceException {
+
+ return new ServerCallable<Boolean>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public Boolean call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+ BlockingInterface tajoMasterService = client.getStub();
+ return tajoMasterService.dropDatabase(null, connection.convertSessionedString(databaseName)).getValue();
+ }
+
+ }.withRetries();
+ }
+
+ @Override
+ public List<String> getAllDatabaseNames() throws ServiceException {
+
+ return new ServerCallable<List<String>>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public List<String> call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+ BlockingInterface tajoMasterService = client.getStub();
+ return tajoMasterService.getAllDatabases(null, connection.sessionId).getValuesList();
+ }
+
+ }.withRetries();
+ }
+
+ public boolean existTable(final String tableName) throws ServiceException {
+
+ return new ServerCallable<Boolean>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public Boolean call(NettyClientBase client) throws ServiceException {
+ connection.checkSessionAndGet(client);
+ BlockingInterface tajoMasterService = client.getStub();
+ return tajoMasterService.existTable(null, connection.convertSessionedString(tableName)).getValue();
+ }
+
+ }.withRetries();
+ }
+
+ @Override
+ public TableDesc createExternalTable(String tableName, Schema schema, Path path, TableMeta meta)
+ throws SQLException, ServiceException {
+ return createExternalTable(tableName, schema, path, meta, null);
+ }
+
+ public TableDesc createExternalTable(final String tableName, final Schema schema, final Path path,
+ final TableMeta meta, final PartitionMethodDesc partitionMethodDesc)
+ throws SQLException, ServiceException {
+
+ return new ServerCallable<TableDesc>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public TableDesc call(NettyClientBase client) throws ServiceException, SQLException {
+
+ connection.checkSessionAndGet(client);
+ BlockingInterface tajoMasterService = client.getStub();
+
+ ClientProtos.CreateTableRequest.Builder builder = ClientProtos.CreateTableRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ builder.setName(tableName);
+ builder.setSchema(schema.getProto());
+ builder.setMeta(meta.getProto());
+ builder.setPath(path.toUri().toString());
+ if (partitionMethodDesc != null) {
+ builder.setPartition(partitionMethodDesc.getProto());
+ }
+ ClientProtos.TableResponse res = tajoMasterService.createExternalTable(null, builder.build());
+ if (res.getResultCode() == ClientProtos.ResultCode.OK) {
+ return CatalogUtil.newTableDesc(res.getTableDesc());
+ } else {
+ throw new SQLException(res.getErrorMessage(), SQLStates.ER_NO_SUCH_TABLE.getState());
+ }
+ }
+
+ }.withRetries();
+ }
+
+ @Override
+ public boolean dropTable(String tableName) throws ServiceException {
+ return dropTable(tableName, false);
+ }
+
+ @Override
+ public boolean dropTable(final String tableName, final boolean purge) throws ServiceException {
+
+ return new ServerCallable<Boolean>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public Boolean call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+ BlockingInterface tajoMasterService = client.getStub();
+
+ ClientProtos.DropTableRequest.Builder builder = ClientProtos.DropTableRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ builder.setName(tableName);
+ builder.setPurge(purge);
+ return tajoMasterService.dropTable(null, builder.build()).getValue();
+ }
+
+ }.withRetries();
+
+ }
+
+ @Override
+ public List<String> getTableList(@Nullable final String databaseName) throws ServiceException {
+ return new ServerCallable<List<String>>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public List<String> call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+ BlockingInterface tajoMasterService = client.getStub();
+
+ ClientProtos.GetTableListRequest.Builder builder = ClientProtos.GetTableListRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ if (databaseName != null) {
+ builder.setDatabaseName(databaseName);
+ }
+ ClientProtos.GetTableListResponse res = tajoMasterService.getTableList(null, builder.build());
+ return res.getTablesList();
+ }
+
+ }.withRetries();
+ }
+
+ @Override
+ public TableDesc getTableDesc(final String tableName) throws ServiceException {
+
+ return new ServerCallable<TableDesc>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public TableDesc call(NettyClientBase client) throws ServiceException, SQLException {
+
+ connection.checkSessionAndGet(client);
+ BlockingInterface tajoMasterService = client.getStub();
+
+ ClientProtos.GetTableDescRequest.Builder builder = ClientProtos.GetTableDescRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ builder.setTableName(tableName);
+ ClientProtos.TableResponse res = tajoMasterService.getTableDesc(null, builder.build());
+ if (res.getResultCode() == ClientProtos.ResultCode.OK) {
+ return CatalogUtil.newTableDesc(res.getTableDesc());
+ } else {
+ throw new SQLException(res.getErrorMessage(), SQLStates.ER_NO_SUCH_TABLE.getState());
+ }
+ }
+
+ }.withRetries();
+ }
+
+ @Override
+ public List<CatalogProtos.FunctionDescProto> getFunctions(final String functionName) throws ServiceException {
+
+ return new ServerCallable<List<CatalogProtos.FunctionDescProto>>(connection.connPool,
+ connection.getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
+
+ public List<CatalogProtos.FunctionDescProto> call(NettyClientBase client) throws ServiceException, SQLException {
+
+ connection.checkSessionAndGet(client);
+ BlockingInterface tajoMasterService = client.getStub();
+
+ String paramFunctionName = functionName == null ? "" : functionName;
+ ClientProtos.FunctionResponse res = tajoMasterService.getFunctionList(null,
+ connection.convertSessionedString(paramFunctionName));
+ if (res.getResultCode() == ClientProtos.ResultCode.OK) {
+ return res.getFunctionsList();
+ } else {
+ throw new SQLException(res.getErrorMessage());
+ }
+ }
+
+ }.withRetries();
+ }
+
+ @Override
+ public void close() throws IOException {
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/InvalidClientSessionException.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/InvalidClientSessionException.java b/tajo-client/src/main/java/org/apache/tajo/client/InvalidClientSessionException.java
new file mode 100644
index 0000000..acbc33f
--- /dev/null
+++ b/tajo-client/src/main/java/org/apache/tajo/client/InvalidClientSessionException.java
@@ -0,0 +1,27 @@
+/**
+ * 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.client;
+
+import com.google.protobuf.ServiceException;
+
+public class InvalidClientSessionException extends ServiceException {
+ public InvalidClientSessionException(String message) {
+ super(message);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/QueryClient.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/QueryClient.java b/tajo-client/src/main/java/org/apache/tajo/client/QueryClient.java
new file mode 100644
index 0000000..dbbafb6
--- /dev/null
+++ b/tajo-client/src/main/java/org/apache/tajo/client/QueryClient.java
@@ -0,0 +1,116 @@
+/**
+ * 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.client;
+
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.tajo.QueryId;
+import org.apache.tajo.ipc.ClientProtos;
+import org.apache.tajo.jdbc.TajoMemoryResultSet;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.sql.ResultSet;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.tajo.TajoIdProtos.SessionIdProto;
+
+public interface QueryClient extends Closeable {
+
+ int UNKNOWN_ROW_NUMBER = -1;
+
+ public void setSessionId(SessionIdProto sessionId);
+
+ public boolean isConnected();
+
+ public SessionIdProto getSessionId();
+
+ public String getBaseDatabase();
+
+ @Override
+ public void close();
+
+ public UserGroupInformation getUserInfo();
+
+ /**
+ * Call to QueryMaster closing query resources
+ * @param queryId
+ */
+ public void closeQuery(final QueryId queryId);
+
+ public void closeNonForwardQuery(final QueryId queryId);
+
+ public String getCurrentDatabase() throws ServiceException;
+
+ public Boolean selectDatabase(final String databaseName) throws ServiceException;
+
+ public Boolean updateSessionVariables(final Map<String, String> variables) throws ServiceException;
+
+ public Boolean unsetSessionVariables(final List<String> variables) throws ServiceException;
+
+ public String getSessionVariable(final String varname) throws ServiceException;
+
+ public Boolean existSessionVariable(final String varname) throws ServiceException;
+
+ public Map<String, String> getAllSessionVariables() throws ServiceException;
+
+ /**
+ * It submits a query statement and get a response immediately.
+ * The response only contains a query id, and submission status.
+ * In order to get the result, you should use {@link #getQueryResult(org.apache.tajo.QueryId)}.
+ */
+ public ClientProtos.SubmitQueryResponse executeQuery(final String sql) throws ServiceException;
+
+ public ClientProtos.SubmitQueryResponse executeQueryWithJson(final String json) throws ServiceException;
+
+ /**
+ * It submits a query statement and get a response.
+ * The main difference from {@link #executeQuery(String)}
+ * is a blocking method. So, this method is wait for
+ * the finish of the submitted query.
+ *
+ * @return If failed, return null.
+ */
+ public ResultSet executeQueryAndGetResult(final String sql) throws ServiceException, IOException;
+
+ public ResultSet executeJsonQueryAndGetResult(final String json) throws ServiceException, IOException;
+
+ public QueryStatus getQueryStatus(QueryId queryId) throws ServiceException;
+
+ public ResultSet getQueryResult(QueryId queryId) throws ServiceException, IOException;
+
+ public ResultSet createNullResultSet(QueryId queryId) throws IOException;
+
+ public ClientProtos.GetQueryResultResponse getResultResponse(QueryId queryId) throws ServiceException;
+
+ public TajoMemoryResultSet fetchNextQueryResult(final QueryId queryId, final int fetchRowNum) throws ServiceException;
+
+ public boolean updateQuery(final String sql) throws ServiceException;
+
+ public boolean updateQueryWithJson(final String json) throws ServiceException;
+
+ public List<ClientProtos.BriefQueryInfo> getRunningQueryList() throws ServiceException;
+
+ public List<ClientProtos.BriefQueryInfo> getFinishedQueryList() throws ServiceException;
+
+ public List<ClientProtos.WorkerResourceInfo> getClusterInfo() throws ServiceException;
+
+ public QueryStatus killQuery(final QueryId queryId) throws ServiceException, IOException;
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java b/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java
new file mode 100644
index 0000000..235ce19
--- /dev/null
+++ b/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java
@@ -0,0 +1,622 @@
+/**
+ * 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.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.QueryId;
+import org.apache.tajo.QueryIdFactory;
+import org.apache.tajo.TajoIdProtos;
+import org.apache.tajo.TajoProtos;
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.ipc.ClientProtos;
+import org.apache.tajo.ipc.QueryMasterClientProtocol;
+import org.apache.tajo.ipc.TajoMasterClientProtocol;
+import org.apache.tajo.jdbc.TajoMemoryResultSet;
+import org.apache.tajo.jdbc.TajoResultSet;
+import org.apache.tajo.rpc.NettyClientBase;
+import org.apache.tajo.rpc.ServerCallable;
+import org.apache.tajo.util.NetUtils;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.sql.ResultSet;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.tajo.conf.TajoConf.ConfVars;
+import static org.apache.tajo.ipc.ClientProtos.*;
+import static org.apache.tajo.ipc.QueryMasterClientProtocol.QueryMasterClientProtocolService;
+import static org.apache.tajo.ipc.TajoMasterClientProtocol.TajoMasterClientProtocolService;
+
+public class QueryClientImpl implements QueryClient {
+ private static final Log LOG = LogFactory.getLog(QueryClientImpl.class);
+ private final SessionConnection connection;
+
+ public QueryClientImpl(SessionConnection connection) {
+ this.connection = connection;
+ }
+
+ @Override
+ public void setSessionId(TajoIdProtos.SessionIdProto sessionId) {
+ connection.setSessionId(sessionId);
+ }
+
+ @Override
+ public boolean isConnected() {
+ return connection.isConnected();
+ }
+
+ @Override
+ public TajoIdProtos.SessionIdProto getSessionId() {
+ return connection.getSessionId();
+ }
+
+ @Override
+ public String getBaseDatabase() {
+ return connection.getBaseDatabase();
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public UserGroupInformation getUserInfo() {
+ return connection.getUserInfo();
+ }
+
+ @Override
+ public void closeQuery(QueryId queryId) {
+ if(connection.queryMasterMap.containsKey(queryId)) {
+ NettyClientBase qmClient = null;
+ try {
+ qmClient = connection.getConnection(queryId, QueryMasterClientProtocol.class, false);
+ QueryMasterClientProtocolService.BlockingInterface queryMaster = qmClient.getStub();
+ queryMaster.closeQuery(null, queryId.getProto());
+ } catch (Exception e) {
+ LOG.warn("Fail to close a QueryMaster connection (qid=" + queryId + ", msg=" + e.getMessage() + ")", e);
+ } finally {
+ connection.connPool.closeConnection(qmClient);
+ connection.queryMasterMap.remove(queryId);
+ }
+ }
+ }
+
+ @Override
+ public void closeNonForwardQuery(QueryId queryId) {
+ NettyClientBase tmClient = null;
+ try {
+ tmClient = connection.getTajoMasterConnection(false);
+ TajoMasterClientProtocolService.BlockingInterface tajoMaster = tmClient.getStub();
+ connection.checkSessionAndGet(tmClient);
+
+ ClientProtos.QueryIdRequest.Builder builder = ClientProtos.QueryIdRequest.newBuilder();
+
+ builder.setSessionId(getSessionId());
+ builder.setQueryId(queryId.getProto());
+ tajoMaster.closeNonForwardQuery(null, builder.build());
+ } catch (Exception e) {
+ LOG.warn("Fail to close a TajoMaster connection (qid=" + queryId + ", msg=" + e.getMessage() + ")", e);
+ } finally {
+ connection.connPool.closeConnection(tmClient);
+ }
+ }
+
+ @Override
+ public String getCurrentDatabase() throws ServiceException {
+ return connection.getCurrentDatabase();
+ }
+
+ @Override
+ public Boolean selectDatabase(String databaseName) throws ServiceException {
+ return connection.selectDatabase(databaseName);
+ }
+
+ @Override
+ public Boolean updateSessionVariables(Map<String, String> variables) throws ServiceException {
+ return connection.updateSessionVariables(variables);
+ }
+
+ @Override
+ public Boolean unsetSessionVariables(List<String> variables) throws ServiceException {
+ return connection.unsetSessionVariables(variables);
+ }
+
+ @Override
+ public String getSessionVariable(String varname) throws ServiceException {
+ return getSessionVariable(varname);
+ }
+
+ @Override
+ public Boolean existSessionVariable(String varname) throws ServiceException {
+ return existSessionVariable(varname);
+ }
+
+ @Override
+ public Map<String, String> getAllSessionVariables() throws ServiceException {
+ return getAllSessionVariables();
+ }
+
+ @Override
+ public ClientProtos.SubmitQueryResponse executeQuery(final String sql) throws ServiceException {
+
+ return new ServerCallable<ClientProtos.SubmitQueryResponse>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public ClientProtos.SubmitQueryResponse call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+
+ final QueryRequest.Builder builder = QueryRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ builder.setQuery(sql);
+ builder.setIsJson(false);
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+
+
+ return tajoMasterService.submitQuery(null, builder.build());
+ }
+ }.withRetries();
+ }
+
+ @Override
+ public ClientProtos.SubmitQueryResponse executeQueryWithJson(final String json) throws ServiceException {
+
+ return new ServerCallable<ClientProtos.SubmitQueryResponse>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public ClientProtos.SubmitQueryResponse call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+
+ final QueryRequest.Builder builder = QueryRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ builder.setQuery(json);
+ builder.setIsJson(true);
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+
+
+ return tajoMasterService.submitQuery(null, builder.build());
+ }
+ }.withRetries();
+ }
+
+ @Override
+ public ResultSet executeQueryAndGetResult(String sql) throws ServiceException, IOException {
+
+ ClientProtos.SubmitQueryResponse response = executeQuery(sql);
+
+ if (response.getResultCode() == ClientProtos.ResultCode.ERROR) {
+ throw new ServiceException(response.getErrorTrace());
+ }
+
+ QueryId queryId = new QueryId(response.getQueryId());
+
+ if (response.getIsForwarded()) {
+ if (queryId.equals(QueryIdFactory.NULL_QUERY_ID)) {
+ return this.createNullResultSet(queryId);
+ } else {
+ return this.getQueryResultAndWait(queryId);
+ }
+
+ } else {
+ // If a non-forwarded insert into query
+ if (queryId.equals(QueryIdFactory.NULL_QUERY_ID) && response.getMaxRowNum() == 0) {
+ return this.createNullResultSet(queryId);
+ } else {
+ if (response.hasResultSet() || response.hasTableDesc()) {
+ return TajoClientUtil.createResultSet(connection.getConf() , this, response);
+ } else {
+ return this.createNullResultSet(queryId);
+ }
+ }
+ }
+ }
+
+ @Override
+ public ResultSet executeJsonQueryAndGetResult(final String json) throws ServiceException, IOException {
+
+ ClientProtos.SubmitQueryResponse response = executeQueryWithJson(json);
+
+ if (response.getResultCode() == ClientProtos.ResultCode.ERROR) {
+ throw new ServiceException(response.getErrorTrace());
+ }
+
+ QueryId queryId = new QueryId(response.getQueryId());
+
+ if (response.getIsForwarded()) {
+
+ if (queryId.equals(QueryIdFactory.NULL_QUERY_ID)) {
+ return this.createNullResultSet(queryId);
+ } else {
+ return this.getQueryResultAndWait(queryId);
+ }
+
+ } else {
+
+ if (response.hasResultSet() || response.hasTableDesc()) {
+ return TajoClientUtil.createResultSet(connection.getConf(), this, response);
+ } else {
+ return this.createNullResultSet(queryId);
+ }
+
+ }
+ }
+
+ private ResultSet getQueryResultAndWait(QueryId queryId) throws ServiceException, IOException {
+
+ if (queryId.equals(QueryIdFactory.NULL_QUERY_ID)) {
+ return createNullResultSet(queryId);
+ }
+
+ QueryStatus status = getQueryStatus(queryId);
+
+ while(status != null && !TajoClientUtil.isQueryComplete(status.getState())) {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ status = getQueryStatus(queryId);
+ }
+
+ if (status.getState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+ if (status.hasResult()) {
+ return getQueryResult(queryId);
+ } else {
+ return createNullResultSet(queryId);
+ }
+
+ } else {
+ LOG.warn("Query (" + status.getQueryId() + ") failed: " + status.getState());
+
+ //TODO throw SQLException(?)
+ return createNullResultSet(queryId);
+ }
+ }
+
+ @Override
+ public QueryStatus getQueryStatus(QueryId queryId) throws ServiceException {
+
+ ClientProtos.GetQueryStatusRequest.Builder builder = ClientProtos.GetQueryStatusRequest.newBuilder();
+ builder.setQueryId(queryId.getProto());
+
+ ClientProtos.GetQueryStatusResponse res = null;
+
+ if(connection.queryMasterMap.containsKey(queryId)) {
+ NettyClientBase qmClient = null;
+
+ try {
+
+ qmClient = connection.connPool.getConnection(connection.queryMasterMap.get(queryId),
+ QueryMasterClientProtocol.class, false);
+ QueryMasterClientProtocolService.BlockingInterface queryMasterService = qmClient.getStub();
+ res = queryMasterService.getQueryStatus(null, builder.build());
+
+ } catch (Exception e) {
+ throw new ServiceException(e.getMessage(), e);
+ } finally {
+ connection.connPool.releaseConnection(qmClient);
+ }
+
+ } else {
+
+ NettyClientBase tmClient = null;
+
+ try {
+ tmClient = connection.getTajoMasterConnection(false);
+ connection.checkSessionAndGet(tmClient);
+ builder.setSessionId(connection.sessionId);
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = tmClient.getStub();
+
+ res = tajoMasterService.getQueryStatus(null, builder.build());
+
+ String queryMasterHost = res.getQueryMasterHost();
+
+ if(queryMasterHost != null && !queryMasterHost.isEmpty()) {
+ NettyClientBase qmClient = null;
+
+ try {
+
+ InetSocketAddress qmAddr = NetUtils.createSocketAddr(queryMasterHost, res.getQueryMasterPort());
+ qmClient = connection.connPool.getConnection(
+ qmAddr, QueryMasterClientProtocol.class, false);
+ QueryMasterClientProtocolService.BlockingInterface queryMasterService = qmClient.getStub();
+ res = queryMasterService.getQueryStatus(null, builder.build());
+
+ connection.queryMasterMap.put(queryId, qmAddr);
+
+ } catch (Exception e) {
+ throw new ServiceException(e.getMessage(), e);
+ } finally {
+ connection.connPool.releaseConnection(qmClient);
+ }
+ }
+
+ } catch (Exception e) {
+ throw new ServiceException(e.getMessage(), e);
+ } finally {
+ connection.connPool.releaseConnection(tmClient);
+ }
+ }
+ return new QueryStatus(res);
+ }
+
+ @Override
+ public ResultSet getQueryResult(QueryId queryId) throws ServiceException, IOException {
+
+ if (queryId.equals(QueryIdFactory.NULL_QUERY_ID)) {
+ return createNullResultSet(queryId);
+ }
+
+ GetQueryResultResponse response = getResultResponse(queryId);
+ TableDesc tableDesc = CatalogUtil.newTableDesc(response.getTableDesc());
+ connection.getConf().setVar(ConfVars.USERNAME, response.getTajoUserName());
+
+ return new TajoResultSet(this, queryId, connection.getConf(), tableDesc);
+ }
+
+ @Override
+ public ResultSet createNullResultSet(QueryId queryId) throws IOException {
+ return new TajoResultSet(this, queryId);
+ }
+
+ @Override
+ public GetQueryResultResponse getResultResponse(QueryId queryId) throws ServiceException {
+ if (queryId.equals(QueryIdFactory.NULL_QUERY_ID)) {
+ return null;
+ }
+
+ NettyClientBase client = null;
+
+ try {
+
+ InetSocketAddress queryMasterAddr = connection.queryMasterMap.get(queryId);
+ if(queryMasterAddr == null) {
+ LOG.warn("No Connection to QueryMaster for " + queryId);
+ return null;
+ }
+
+ client = connection.getConnection(queryMasterAddr, QueryMasterClientProtocol.class, false);
+ QueryMasterClientProtocolService.BlockingInterface queryMasterService = client.getStub();
+
+ GetQueryResultRequest.Builder builder = GetQueryResultRequest.newBuilder();
+ builder.setQueryId(queryId.getProto());
+ GetQueryResultResponse response = queryMasterService.getQueryResult(null,builder.build());
+
+ return response;
+
+ } catch (Exception e) {
+ throw new ServiceException(e.getMessage(), e);
+ } finally {
+ connection.connPool.releaseConnection(client);
+ }
+ }
+
+ @Override
+ public TajoMemoryResultSet fetchNextQueryResult(final QueryId queryId, final int fetchRowNum)
+ throws ServiceException {
+
+ try {
+ ServerCallable<ClientProtos.SerializedResultSet> callable =
+ new ServerCallable<ClientProtos.SerializedResultSet>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public ClientProtos.SerializedResultSet call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+
+ GetQueryResultDataRequest.Builder builder = GetQueryResultDataRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ builder.setQueryId(queryId.getProto());
+ builder.setFetchRowNum(fetchRowNum);
+ try {
+ GetQueryResultDataResponse response = tajoMasterService.getQueryResultData(null, builder.build());
+ if (response.getResultCode() == ClientProtos.ResultCode.ERROR) {
+ throw new ServiceException(response.getErrorTrace());
+ }
+
+ return response.getResultSet();
+ } catch (ServiceException e) {
+ abort();
+ throw e;
+ } catch (Throwable t) {
+ throw new ServiceException(t.getMessage(), t);
+ }
+ }
+ };
+
+ ClientProtos.SerializedResultSet serializedResultSet = callable.withRetries();
+
+ return new TajoMemoryResultSet(
+ new Schema(serializedResultSet.getSchema()),
+ serializedResultSet.getSerializedTuplesList(),
+ serializedResultSet.getSerializedTuplesCount());
+ } catch (Exception e) {
+ throw new ServiceException(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public boolean updateQuery(final String sql) throws ServiceException {
+
+ return new ServerCallable<Boolean>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public Boolean call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+
+ QueryRequest.Builder builder = QueryRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ builder.setQuery(sql);
+ builder.setIsJson(false);
+ ClientProtos.UpdateQueryResponse response = tajoMasterService.updateQuery(null, builder.build());
+
+ if (response.getResultCode() == ClientProtos.ResultCode.OK) {
+ return true;
+ } else {
+ if (response.hasErrorMessage()) {
+ System.err.println("ERROR: " + response.getErrorMessage());
+ }
+ return false;
+ }
+ }
+ }.withRetries();
+ }
+
+ @Override
+ public boolean updateQueryWithJson(final String json) throws ServiceException {
+
+ return new ServerCallable<Boolean>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public Boolean call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+
+ QueryRequest.Builder builder = QueryRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ builder.setQuery(json);
+ builder.setIsJson(true);
+ ClientProtos.UpdateQueryResponse response = tajoMasterService.updateQuery(null, builder.build());
+ if (response.getResultCode() == ClientProtos.ResultCode.OK) {
+ return true;
+ } else {
+ if (response.hasErrorMessage()) {
+ System.err.println("ERROR: " + response.getErrorMessage());
+ }
+ return false;
+ }
+ }
+ }.withRetries();
+ }
+
+ @Override
+ public List<ClientProtos.BriefQueryInfo> getRunningQueryList() throws ServiceException {
+
+ return new ServerCallable<List<ClientProtos.BriefQueryInfo>>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public List<ClientProtos.BriefQueryInfo> call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+
+ ClientProtos.GetQueryListRequest.Builder builder = ClientProtos.GetQueryListRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ ClientProtos.GetQueryListResponse res = tajoMasterService.getRunningQueryList(null, builder.build());
+ return res.getQueryListList();
+
+ }
+ }.withRetries();
+ }
+
+ @Override
+ public List<ClientProtos.BriefQueryInfo> getFinishedQueryList() throws ServiceException {
+
+ return new ServerCallable<List<ClientProtos.BriefQueryInfo>>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public List<ClientProtos.BriefQueryInfo> call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+
+ ClientProtos.GetQueryListRequest.Builder builder = ClientProtos.GetQueryListRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ ClientProtos.GetQueryListResponse res = tajoMasterService.getFinishedQueryList(null, builder.build());
+ return res.getQueryListList();
+
+ }
+ }.withRetries();
+ }
+
+ @Override
+ public List<ClientProtos.WorkerResourceInfo> getClusterInfo() throws ServiceException {
+
+ return new ServerCallable<List<ClientProtos.WorkerResourceInfo>>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+
+ public List<ClientProtos.WorkerResourceInfo> call(NettyClientBase client) throws ServiceException {
+
+ connection.checkSessionAndGet(client);
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+
+ ClientProtos.GetClusterInfoRequest.Builder builder = ClientProtos.GetClusterInfoRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ ClientProtos.GetClusterInfoResponse res = tajoMasterService.getClusterInfo(null, builder.build());
+ return res.getWorkerListList();
+ }
+
+ }.withRetries();
+ }
+
+ @Override
+ public QueryStatus killQuery(final QueryId queryId)
+ throws ServiceException, IOException {
+
+ QueryStatus status = getQueryStatus(queryId);
+
+ NettyClientBase tmClient = null;
+ try {
+ /* send a kill to the TM */
+ tmClient = connection.getTajoMasterConnection(false);
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = tmClient.getStub();
+
+ connection.checkSessionAndGet(tmClient);
+
+ ClientProtos.QueryIdRequest.Builder builder = ClientProtos.QueryIdRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ builder.setQueryId(queryId.getProto());
+ tajoMasterService.killQuery(null, builder.build());
+
+ long currentTimeMillis = System.currentTimeMillis();
+ long timeKillIssued = currentTimeMillis;
+ while ((currentTimeMillis < timeKillIssued + 10000L)
+ && ((status.getState() != TajoProtos.QueryState.QUERY_KILLED)
+ || (status.getState() == TajoProtos.QueryState.QUERY_KILL_WAIT))) {
+ try {
+ Thread.sleep(100L);
+ } catch(InterruptedException ie) {
+ break;
+ }
+ currentTimeMillis = System.currentTimeMillis();
+ status = getQueryStatus(queryId);
+ }
+
+ } catch(Exception e) {
+ LOG.debug("Error when checking for application status", e);
+ } finally {
+ connection.connPool.releaseConnection(tmClient);
+ }
+ return status;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java b/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java
new file mode 100644
index 0000000..42085a2
--- /dev/null
+++ b/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java
@@ -0,0 +1,326 @@
+/**
+ * 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.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.QueryId;
+import org.apache.tajo.TajoIdProtos;
+import org.apache.tajo.annotation.Nullable;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.ipc.ClientProtos;
+import org.apache.tajo.ipc.TajoMasterClientProtocol;
+import org.apache.tajo.rpc.NettyClientBase;
+import org.apache.tajo.rpc.RpcConnectionPool;
+import org.apache.tajo.rpc.ServerCallable;
+import org.apache.tajo.util.HAServiceUtil;
+import org.apache.tajo.util.KeyValueSet;
+import org.apache.tajo.util.NetUtils;
+import org.jboss.netty.channel.ConnectTimeoutException;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.tajo.ipc.ClientProtos.CreateSessionRequest;
+import static org.apache.tajo.ipc.ClientProtos.CreateSessionResponse;
+import static org.apache.tajo.ipc.TajoMasterClientProtocol.TajoMasterClientProtocolService;
+
+public class SessionConnection implements Closeable {
+
+ private final Log LOG = LogFactory.getLog(TajoClientImpl.class);
+
+ private final TajoConf conf;
+
+ final Map<QueryId, InetSocketAddress> queryMasterMap = new ConcurrentHashMap<QueryId, InetSocketAddress>();
+
+ final InetSocketAddress tajoMasterAddr;
+
+ final RpcConnectionPool connPool;
+
+ private final String baseDatabase;
+
+ private final UserGroupInformation userInfo;
+
+ volatile TajoIdProtos.SessionIdProto sessionId;
+
+ private AtomicBoolean closed = new AtomicBoolean(false);
+
+
+ public SessionConnection(TajoConf conf) throws IOException {
+ this(conf, NetUtils.createSocketAddr(conf.getVar(TajoConf.ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS)), null);
+ }
+
+ public SessionConnection(TajoConf conf, @Nullable String baseDatabase) throws IOException {
+ this(conf, NetUtils.createSocketAddr(conf.getVar(TajoConf.ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS)), baseDatabase);
+ }
+
+ public SessionConnection(InetSocketAddress addr) throws IOException {
+ this(new TajoConf(), addr, null);
+ }
+
+ public SessionConnection(String hostname, int port, String baseDatabase) throws IOException {
+ this(new TajoConf(), NetUtils.createSocketAddr(hostname, port), baseDatabase);
+ }
+
+ /**
+ * Connect to TajoMaster
+ *
+ * @param conf TajoConf
+ * @param addr TajoMaster address
+ * @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 SessionConnection(TajoConf conf, InetSocketAddress addr, @Nullable String baseDatabase) throws IOException {
+ this.conf = conf;
+ this.conf.set("tajo.disk.scheduler.report.interval", "0");
+ this.tajoMasterAddr = addr;
+ int workerNum = conf.getIntVar(TajoConf.ConfVars.RPC_CLIENT_WORKER_THREAD_NUM);
+ // Don't share connection pool per client
+ connPool = RpcConnectionPool.newPool(conf, getClass().getSimpleName(), workerNum);
+ userInfo = UserGroupInformation.getCurrentUser();
+ this.baseDatabase = baseDatabase != null ? baseDatabase : null;
+ }
+
+ public <T> T getStub(QueryId queryId, Class protocolClass, boolean asyncMode) throws NoSuchMethodException,
+ ConnectTimeoutException, ClassNotFoundException {
+ InetSocketAddress addr = queryMasterMap.get(queryId);
+ return connPool.getConnection(addr, protocolClass, asyncMode).getStub();
+ }
+
+ public NettyClientBase getTajoMasterConnection(boolean asyncMode) throws NoSuchMethodException,
+ ConnectTimeoutException, ClassNotFoundException {
+ return connPool.getConnection(getTajoMasterAddr(), TajoMasterClientProtocol.class, asyncMode);
+ }
+
+ public NettyClientBase getConnection(QueryId queryId, Class protocolClass, boolean asyncMode)
+ throws NoSuchMethodException, ConnectTimeoutException, ClassNotFoundException {
+ InetSocketAddress addr = queryMasterMap.get(queryId);
+ return connPool.getConnection(addr, protocolClass, asyncMode);
+ }
+
+ public NettyClientBase getConnection(InetSocketAddress addr, Class protocolClass, boolean asyncMode)
+ throws NoSuchMethodException, ConnectTimeoutException, ClassNotFoundException {
+ return connPool.getConnection(addr, protocolClass, asyncMode);
+ }
+
+ @SuppressWarnings("unused")
+ public void setSessionId(TajoIdProtos.SessionIdProto sessionId) {
+ this.sessionId = sessionId;
+ }
+
+ public TajoIdProtos.SessionIdProto getSessionId() {
+ return sessionId;
+ }
+
+ public String getBaseDatabase() {
+ return baseDatabase;
+ }
+
+ public boolean isConnected() {
+ if(!closed.get()){
+ try {
+ return connPool.getConnection(tajoMasterAddr, TajoMasterClientProtocol.class, false).isConnected();
+ } catch (Throwable e) {
+ return false;
+ }
+ }
+ return false;
+ }
+
+ public TajoConf getConf() {
+ return conf;
+ }
+
+ public UserGroupInformation getUserInfo() {
+ return userInfo;
+ }
+
+ public String getCurrentDatabase() throws ServiceException {
+ return new ServerCallable<String>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
+
+ public String call(NettyClientBase client) throws ServiceException {
+ checkSessionAndGet(client);
+
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+ return tajoMasterService.getCurrentDatabase(null, sessionId).getValue();
+ }
+ }.withRetries();
+ }
+
+ public Boolean updateSessionVariables(final Map<String, String> variables) throws ServiceException {
+ return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
+
+ public Boolean call(NettyClientBase client) throws ServiceException {
+ checkSessionAndGet(client);
+
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+ KeyValueSet keyValueSet = new KeyValueSet();
+ keyValueSet.putAll(variables);
+ ClientProtos.UpdateSessionVariableRequest request = ClientProtos.UpdateSessionVariableRequest.newBuilder()
+ .setSessionId(sessionId)
+ .setSetVariables(keyValueSet.getProto()).build();
+
+ return tajoMasterService.updateSessionVariables(null, request).getValue();
+ }
+ }.withRetries();
+ }
+
+ public Boolean unsetSessionVariables(final List<String> variables) throws ServiceException {
+ return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
+
+ public Boolean call(NettyClientBase client) throws ServiceException {
+ checkSessionAndGet(client);
+
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+ ClientProtos.UpdateSessionVariableRequest request = ClientProtos.UpdateSessionVariableRequest.newBuilder()
+ .setSessionId(sessionId)
+ .addAllUnsetVariables(variables).build();
+ return tajoMasterService.updateSessionVariables(null, request).getValue();
+ }
+ }.withRetries();
+ }
+
+ public String getSessionVariable(final String varname) throws ServiceException {
+ return new ServerCallable<String>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
+
+ public String call(NettyClientBase client) throws ServiceException {
+ checkSessionAndGet(client);
+
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+ return tajoMasterService.getSessionVariable(null, convertSessionedString(varname)).getValue();
+ }
+ }.withRetries();
+ }
+
+ public Boolean existSessionVariable(final String varname) throws ServiceException {
+ return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
+
+ public Boolean call(NettyClientBase client) throws ServiceException {
+ checkSessionAndGet(client);
+
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+ return tajoMasterService.existSessionVariable(null, convertSessionedString(varname)).getValue();
+ }
+ }.withRetries();
+ }
+
+ public Map<String, String> getAllSessionVariables() throws ServiceException {
+ return new ServerCallable<Map<String, String>>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class,
+ false, true) {
+
+ public Map<String, String> call(NettyClientBase client) throws ServiceException {
+ checkSessionAndGet(client);
+
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+ KeyValueSet keyValueSet = new KeyValueSet(tajoMasterService.getAllSessionVariables(null, sessionId));
+ return keyValueSet.getAllKeyValus();
+ }
+ }.withRetries();
+ }
+
+ public Boolean selectDatabase(final String databaseName) throws ServiceException {
+ return new ServerCallable<Boolean>(connPool, getTajoMasterAddr(), TajoMasterClientProtocol.class, false, true) {
+
+ public Boolean call(NettyClientBase client) throws ServiceException {
+ checkSessionAndGet(client);
+
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+ return tajoMasterService.selectDatabase(null, convertSessionedString(databaseName)).getValue();
+ }
+ }.withRetries();
+ }
+
+ @Override
+ public void close() {
+ if(closed.getAndSet(true)){
+ return;
+ }
+
+ // remove session
+ try {
+
+ NettyClientBase client = connPool.getConnection(getTajoMasterAddr(), TajoMasterClientProtocol.class, false);
+ TajoMasterClientProtocolService.BlockingInterface tajoMaster = client.getStub();
+ tajoMaster.removeSession(null, sessionId);
+
+ } catch (Throwable e) {
+ }
+
+ if(connPool != null) {
+ connPool.shutdown();
+ }
+
+ queryMasterMap.clear();
+ }
+
+ protected InetSocketAddress getTajoMasterAddr() {
+ if (!conf.getBoolVar(TajoConf.ConfVars.TAJO_MASTER_HA_ENABLE)) {
+ return tajoMasterAddr;
+ } else {
+ if (!HAServiceUtil.isMasterAlive(tajoMasterAddr, conf)) {
+ return HAServiceUtil.getMasterClientAddress(conf);
+ } else {
+ return tajoMasterAddr;
+ }
+ }
+ }
+
+ protected void checkSessionAndGet(NettyClientBase client) throws ServiceException {
+
+ if (sessionId == null) {
+
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+ CreateSessionRequest.Builder builder = CreateSessionRequest.newBuilder();
+ builder.setUsername(userInfo.getUserName()).build();
+
+ if (baseDatabase != null) {
+ builder.setBaseDatabaseName(baseDatabase);
+ }
+
+ CreateSessionResponse response = tajoMasterService.createSession(null, builder.build());
+
+ if (response.getState() == CreateSessionResponse.ResultState.SUCCESS) {
+
+ sessionId = response.getSessionId();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(String.format("Got session %s as a user '%s'.", sessionId.getId(), userInfo.getUserName()));
+ }
+
+ } else {
+ throw new InvalidClientSessionException(response.getMessage());
+ }
+ }
+ }
+
+ ClientProtos.SessionedStringProto convertSessionedString(String str) {
+ ClientProtos.SessionedStringProto.Builder builder = ClientProtos.SessionedStringProto.newBuilder();
+ builder.setSessionId(sessionId);
+ builder.setValue(str);
+ return builder.build();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/f80beaf6/tajo-client/src/main/java/org/apache/tajo/client/TajoAdmin.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/TajoAdmin.java b/tajo-client/src/main/java/org/apache/tajo/client/TajoAdmin.java
index 498402c..817a698 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/TajoAdmin.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/TajoAdmin.java
@@ -143,9 +143,9 @@ public class TajoAdmin {
return;
} else if (hostName != null && port != null) {
tajoConf.setVar(TajoConf.ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS, hostName + ":" + port);
- tajoClient = new TajoClient(tajoConf);
+ tajoClient = new TajoClientImpl(tajoConf);
} else if (hostName == null && port == null) {
- tajoClient = new TajoClient(tajoConf);
+ tajoClient = new TajoClientImpl(tajoConf);
}
switch (cmdType) {
@@ -196,7 +196,7 @@ public class TajoAdmin {
long end = queryInfo.getFinishTime();
long start = queryInfo.getStartTime();
String executionTime = decimalF.format((end-start) / 1000) + " sec";
- if (TajoClient.isInCompleteState(queryInfo.getState())) {
+ if (TajoClientUtil.isQueryComplete(queryInfo.getState())) {
writer.write("Finished Time: " + df.format(queryInfo.getFinishTime()));
writer.write("\n");
}
@@ -434,7 +434,7 @@ public class TajoAdmin {
}
writer.write("\n");
} else {
- String confMasterServiceAddr = tajoClient.getConf().getVar(TajoConf.ConfVars.TAJO_MASTER_UMBILICAL_RPC_ADDRESS);
+ String confMasterServiceAddr = tajoConf.getVar(TajoConf.ConfVars.TAJO_MASTER_UMBILICAL_RPC_ADDRESS);
InetSocketAddress masterAddress = NetUtils.createSocketAddr(confMasterServiceAddr);
writer.write(masterAddress.getHostName());
writer.write("\n");