[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[parthchandra]

Github user parthchandra commented on the pull request:

https://github.com/apache/drill/pull/396#issuecomment-193864509
  
Merged in df0f0af3d963c1b65eb01c3141fe84532c53f5a5


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[parthchandra]

Github user parthchandra closed the pull request at:

https://github.com/apache/drill/pull/396


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[hnfgns]

Github user hnfgns commented on the pull request:

https://github.com/apache/drill/pull/396#issuecomment-193499159
  
+1 once comments above are addressed.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[hnfgns]

Github user hnfgns commented on a diff in the pull request:

https://github.com/apache/drill/pull/396#discussion_r55290473
  
--- Diff: contrib/native/client/src/clientlib/drillClientImpl.cpp ---
@@ -1392,6 +1390,206 @@ void DrillClientQueryResult::clearAndDestroy(){
 }
 }
 
+
+connectionStatus_t PooledDrillClientImpl::connect(const char* connStr){
+connectionStatus_t stat = CONN_SUCCESS;
+std::string pathToDrill, protocol, hostPortStr;
+std::string host;
+std::string port;
+m_connectStr=connStr;
+Utils::parseConnectStr(connStr, pathToDrill, protocol, hostPortStr);
+if(!strcmp(protocol.c_str(), "zk")){
+// Get a list of drillbits
+ZookeeperImpl zook;
+std::vector drillbits;
+int err = zook.getAllDrillbits(hostPortStr.c_str(), 
pathToDrill.c_str(), drillbits);
+if(!err){
+Utils::shuffle(drillbits);
+// The original shuffled order is maintained if we shuffle 
first and then add any missing elements
+Utils::add(m_drillbits, drillbits);
+exec::DrillbitEndpoint e;
+size_t nextIndex=0;
+{
+boost::lock_guard cLock(m_cMutex);
+m_lastConnection++;
+nextIndex = (m_lastConnection)%(getDrillbitCount());
+}
+DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Pooled Connection"
+<< "(" << (void*)this << ")"
+<< ": Current counter is: " 
+<< m_lastConnection << std::endl;)
+err=zook.getEndPoint(m_drillbits, nextIndex, e);
+if(!err){
+host=boost::lexical_cast(e.address());
+port=boost::lexical_cast(e.user_port());
+}
+}
+if(err){
+return handleConnError(CONN_ZOOKEEPER_ERROR, 
getMessage(ERR_CONN_ZOOKEEPER, zook.getError().c_str()));
+}
+zook.close();
+m_bIsDirectConnection=false;
+}else if(!strcmp(protocol.c_str(), "local")){
+char tempStr[MAX_CONNECT_STR+1];
+strncpy(tempStr, hostPortStr.c_str(), MAX_CONNECT_STR); 
tempStr[MAX_CONNECT_STR]=0;
+host=strtok(tempStr, ":");
+port=strtok(NULL, "");
+m_bIsDirectConnection=true;
+}else{
+return handleConnError(CONN_INVALID_INPUT, 
getMessage(ERR_CONN_UNKPROTO, protocol.c_str()));
+}
+DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Connecting to endpoint: (Pooled) 
" << host << ":" << port << std::endl;)
+DrillClientImpl* pDrillClientImpl = new DrillClientImpl();
+stat =  pDrillClientImpl->connect(host.c_str(), port.c_str());
+if(stat == CONN_SUCCESS){
+boost::lock_guard lock(m_poolMutex);
+m_clientConnections.push_back(pDrillClientImpl);
+}else{
+DrillClientError* pErr = pDrillClientImpl->getError();
+handleConnError((connectionStatus_t)pErr->status, pErr->msg);
+delete pDrillClientImpl;
+}
+return stat;
+}
+
+connectionStatus_t 
PooledDrillClientImpl::validateHandshake(DrillUserProperties* props){
+// Assume there is one valid connection to at least one drillbit
+connectionStatus_t stat=CONN_FAILURE;
+// Keep a copy of the user properties
+if(props!=NULL){
+m_pUserProperties = new DrillUserProperties;
+for(size_t i=0; isize(); i++){
+m_pUserProperties->setProperty(
+props->keyAt(i),
+props->valueAt(i)
+);
+}
+}
+DrillClientImpl* pDrillClientImpl = getOneConnection();
+if(pDrillClientImpl != NULL){
+DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Validating handshake: 
(Pooled) " << pDrillClientImpl->m_connectedHost << std::endl;)
+stat=pDrillClientImpl->validateHandshake(m_pUserProperties);
+}
+else{
+stat =  handleConnError(CONN_NOTCONNECTED, 
getMessage(ERR_CONN_NOCONN));
+}
+return stat;
+}
+
+DrillClientQueryResult* 
PooledDrillClientImpl::SubmitQuery(::exec::shared::QueryType t, const 
std::string& plan, pfnQueryResultsListener listener, void* listenerCtx){
+DrillClientQueryResult* pDrillClientQueryResult = NULL;
+DrillClientImpl* pDrillClientImpl = NULL;
+pDrillClientImpl = getOneConnection();
+if(pDrillClientImpl != NULL){
+
pDrillClientQueryResult=pDrillClientImpl->SubmitQuery(t,plan,listener,listenerCtx);
+m_queriesExecuted++;
+}
+return pDrillClientQueryResult;
+}
+
+void PooledDrillClientImpl::freeQueryResources(DrillClientQueryResult* 
pQryResult){
+// 

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[hnfgns]

Github user hnfgns commented on a diff in the pull request:

https://github.com/apache/drill/pull/396#discussion_r55286410
  
--- Diff: contrib/native/client/example/querySubmitter.cpp ---
@@ -317,16 +320,20 @@ int main(int argc, char* argv[]) {
 std::vector::iterator queryHandleIter;
 
 Drill::DrillClient client;
-// To log to file
-//DrillClient::initLogging("/var/log/drill/", l);
+#if defined _WIN32 || defined _WIN64
+   const char* logpathPrefix = 
"C:\\Users\\Administrator\\Documents\\temp\\drillclient";
--- End diff --

minor: the use of GetTempPath in  seems more appropriate here.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[paul-rogers]

Github user paul-rogers commented on the pull request:

https://github.com/apache/drill/pull/396#issuecomment-193489132
  
Thanks for addressing the review comments!
+1



---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[parthchandra]

Github user parthchandra commented on the pull request:

https://github.com/apache/drill/pull/396#issuecomment-192797744
  
Updated the PR to address review comments.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[parthchandra]

Github user parthchandra commented on a diff in the pull request:

https://github.com/apache/drill/pull/396#discussion_r55117740
  
--- Diff: contrib/native/client/example/pooledConnections.cpp ---
@@ -0,0 +1,300 @@
+/*
+ * 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.
+ */
+
+#include 
+#include 
+#include 
+#include 
+#include 
+#include "drill/drillc.hpp"
+
+int nOptions=5;
+
+struct Option{
+char name[32];
+char desc[128];
+bool required;
+}qsOptions[]= {
+{"query", "Query strings, separated by semicolons", true},
+{"connectStr", "Connect string", true},
+{"logLevel", "Logging level [trace|debug|info|warn|error|fatal]", 
false},
+{"numConnections", "Number of simultaneous connections", true},
+{"numIterations", "Number of iterations to run. Each query is sent to 
each connection this many times", true}
+};
+
+std::map qsOptionValues;
+
+const char* exceptionInject="alter session set 
`drill.exec.testing.controls` = '{ \"injections\" : [{ 
\"type\":\"exception\",\"siteClass\":\"org.apache.drill.exec.work.fragment.FragmentExecutor\",\"desc\":\"fragment-execution\",\"nSkip\":0,\"nFire\":1,\"exceptionClass\":\"java.lang.OutOfMemoryError\"}]}'";
--- End diff --

Actually, I'm going to remove this example. It is not correct as it depends 
on setting session options which do not work for pooled connections.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[parthchandra]

Github user parthchandra commented on a diff in the pull request:

https://github.com/apache/drill/pull/396#discussion_r55117425
  
--- Diff: contrib/native/client/src/clientlib/drillClientImpl.cpp ---
@@ -1392,6 +1387,198 @@ void DrillClientQueryResult::clearAndDestroy(){
 }
 }
 
+
+connectionStatus_t PooledDrillClientImpl::connect(const char* connStr){
+connectionStatus_t stat = CONN_SUCCESS;
+std::string pathToDrill, protocol, hostPortStr;
+std::string host;
+std::string port;
+m_connectStr=connStr;
+Utils::parseConnectStr(connStr, pathToDrill, protocol, hostPortStr);
+if(!strcmp(protocol.c_str(), "zk")){
+// Get a list of drillbits
+ZookeeperImpl zook;
+std::vector drillbits;
+int err = zook.getAllDrillbits(hostPortStr.c_str(), 
pathToDrill.c_str(), drillbits);
+if(!err){
+Utils::shuffle(drillbits);
+// The original shuffled order is maintained if we shuffle 
first and then add any missing elements
+Utils::add(m_drillbits, drillbits);
+exec::DrillbitEndpoint e;
+size_t nextIndex=0;
+boost::lock_guard cLock(m_cMutex);
+m_lastConnection++;
+nextIndex = (m_lastConnection)%(getDrillbitCount());
+DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Pooled Connection"
+<< "(" << (void*)this << ")"
+<< ": Current counter is: " 
+<< m_lastConnection << std::endl;)
+err=zook.getEndPoint(m_drillbits, nextIndex, e);
+if(!err){
+host=boost::lexical_cast(e.address());
+port=boost::lexical_cast(e.user_port());
+}
+}
+if(err){
+return handleConnError(CONN_ZOOKEEPER_ERROR, 
getMessage(ERR_CONN_ZOOKEEPER, zook.getError().c_str()));
+}
+zook.close();
+m_bIsDirectConnection=false;
+}else if(!strcmp(protocol.c_str(), "local")){
+char tempStr[MAX_CONNECT_STR+1];
+strncpy(tempStr, hostPortStr.c_str(), MAX_CONNECT_STR); 
tempStr[MAX_CONNECT_STR]=0;
+host=strtok(tempStr, ":");
+port=strtok(NULL, "");
+m_bIsDirectConnection=true;
+}else{
+return handleConnError(CONN_INVALID_INPUT, 
getMessage(ERR_CONN_UNKPROTO, protocol.c_str()));
+}
+DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Connecting to endpoint: (Pooled) 
" << host << ":" << port << std::endl;)
+DrillClientImpl* pDrillClientImpl = new DrillClientImpl();
+stat =  pDrillClientImpl->connect(host.c_str(), port.c_str());
+if(stat == CONN_SUCCESS){
+m_clientConnections.push_back(pDrillClientImpl);
+}else{
+DrillClientError* pErr = pDrillClientImpl->getError();
+handleConnError((connectionStatus_t)pErr->status, pErr->msg);
+delete pDrillClientImpl;
+}
+return stat;
+}
+
+connectionStatus_t 
PooledDrillClientImpl::validateHandshake(DrillUserProperties* props){
+// Assume there is one valid connection to at least one drillbit
+connectionStatus_t stat=CONN_FAILURE;
+// Keep a copy of the user properties
+if(props!=NULL){
+m_pUserProperties = new DrillUserProperties;
+for(size_t i=0; isize(); i++){
+m_pUserProperties->setProperty(
+props->keyAt(i),
+props->valueAt(i)
+);
+}
+}
+DrillClientImpl* pDrillClientImpl = getOneConnection();
+if(pDrillClientImpl != NULL){
+DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Validating handshake: 
(Pooled) " << pDrillClientImpl->m_connectedHost << std::endl;)
+stat=pDrillClientImpl->validateHandshake(m_pUserProperties);
+}
+else{
+stat =  handleConnError(CONN_NOTCONNECTED, 
getMessage(ERR_CONN_NOCONN));
+}
+return stat;
+}
+
+DrillClientQueryResult* 
PooledDrillClientImpl::SubmitQuery(::exec::shared::QueryType t, const 
std::string& plan, pfnQueryResultsListener listener, void* listenerCtx){
+DrillClientQueryResult* pDrillClientQueryResult = NULL;
+DrillClientImpl* pDrillClientImpl = NULL;
+pDrillClientImpl = getOneConnection();
+if(pDrillClientImpl != NULL){
+
pDrillClientQueryResult=pDrillClientImpl->SubmitQuery(t,plan,listener,listenerCtx);
+m_queriesExecuted++;
+}
+return pDrillClientQueryResult;
+}
+
+void PooledDrillClientImpl::freeQueryResources(DrillClientQueryResult* 
pQryResult){
+// Nothing to do. If this class ever keeps track of executing queries 
then it will need 
+// 

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[parthchandra]

Github user parthchandra commented on a diff in the pull request:

https://github.com/apache/drill/pull/396#discussion_r55117339
  
--- Diff: contrib/native/client/example/pooledConnections.cpp ---
@@ -0,0 +1,300 @@
+/*
+ * 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.
+ */
+
+#include 
+#include 
+#include 
+#include 
+#include 
+#include "drill/drillc.hpp"
+
+int nOptions=5;
+
+struct Option{
+char name[32];
+char desc[128];
+bool required;
+}qsOptions[]= {
+{"query", "Query strings, separated by semicolons", true},
+{"connectStr", "Connect string", true},
+{"logLevel", "Logging level [trace|debug|info|warn|error|fatal]", 
false},
+{"numConnections", "Number of simultaneous connections", true},
+{"numIterations", "Number of iterations to run. Each query is sent to 
each connection this many times", true}
+};
+
+std::map qsOptionValues;
+
+const char* exceptionInject="alter session set 
`drill.exec.testing.controls` = '{ \"injections\" : [{ 
\"type\":\"exception\",\"siteClass\":\"org.apache.drill.exec.work.fragment.FragmentExecutor\",\"desc\":\"fragment-execution\",\"nSkip\":0,\"nFire\":1,\"exceptionClass\":\"java.lang.OutOfMemoryError\"}]}'";
+
+Drill::status_t SchemaListener(void* ctx, Drill::FieldDefPtr fields, 
Drill::DrillClientError* err){
+if(!err){
+printf("SCHEMA CHANGE DETECTED:\n");
+for(size_t i=0; isize(); i++){
+std::string name= fields->at(i)->getName();
+printf("%s\t", name.c_str());
+}
+printf("\n");
+return Drill::QRY_SUCCESS ;
+}else{
+std::cerr<< "ERROR: " << err->msg << std::endl;
+return Drill::QRY_FAILURE;
+}
+}
+
+boost::mutex listenerMutex;
+Drill::status_t QueryResultsListener(void* ctx, Drill::RecordBatch* b, 
Drill::DrillClientError* err){
+boost::lock_guard listenerLock(listenerMutex);
+if(!err){
+if(b!=NULL){
+b->print(std::cout, 0); // print all rows
+std::cout << "DATA RECEIVED ..." << std::endl;
+delete b; // we're done with this batch, we can delete it
+return Drill::QRY_FAILURE;
+}else{
+std::cout << "Query Complete." << std::endl;
+return Drill::QRY_SUCCESS;
+   }
+}else{
+assert(b==NULL);
+switch(err->status) {
+case Drill::QRY_COMPLETED:
+case Drill::QRY_CANCELED:
+std::cerr<< "INFO: " << err->msg << std::endl;
+return Drill::QRY_SUCCESS;
--- End diff --

Yes. Bit of a bad design there. The return value of failure is actually to 
signal that the listener that the query completed. WE added another api to 
check the status explicitly later. Unfortunately, the ODBC driver now uses this 
so hard to change.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[parthchandra]

Github user parthchandra commented on a diff in the pull request:

https://github.com/apache/drill/pull/396#discussion_r55117302
  
--- Diff: contrib/native/client/example/pooledConnections.cpp ---
@@ -0,0 +1,300 @@
+/*
+ * 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.
+ */
+
+#include 
+#include 
+#include 
+#include 
+#include 
+#include "drill/drillc.hpp"
+
+int nOptions=5;
+
+struct Option{
+char name[32];
+char desc[128];
+bool required;
+}qsOptions[]= {
+{"query", "Query strings, separated by semicolons", true},
+{"connectStr", "Connect string", true},
+{"logLevel", "Logging level [trace|debug|info|warn|error|fatal]", 
false},
+{"numConnections", "Number of simultaneous connections", true},
+{"numIterations", "Number of iterations to run. Each query is sent to 
each connection this many times", true}
+};
+
+std::map qsOptionValues;
+
+const char* exceptionInject="alter session set 
`drill.exec.testing.controls` = '{ \"injections\" : [{ 
\"type\":\"exception\",\"siteClass\":\"org.apache.drill.exec.work.fragment.FragmentExecutor\",\"desc\":\"fragment-execution\",\"nSkip\":0,\"nFire\":1,\"exceptionClass\":\"java.lang.OutOfMemoryError\"}]}'";
--- End diff --

What do i need to add?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[sudheeshkatkam]

Github user sudheeshkatkam commented on a diff in the pull request:

https://github.com/apache/drill/pull/396#discussion_r55112200
  
--- Diff: contrib/native/client/example/pooledConnections.cpp ---
@@ -0,0 +1,300 @@
+/*
+ * 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.
+ */
+
+#include 
+#include 
+#include 
+#include 
+#include 
+#include "drill/drillc.hpp"
+
+int nOptions=5;
+
+struct Option{
+char name[32];
+char desc[128];
+bool required;
+}qsOptions[]= {
+{"query", "Query strings, separated by semicolons", true},
+{"connectStr", "Connect string", true},
+{"logLevel", "Logging level [trace|debug|info|warn|error|fatal]", 
false},
+{"numConnections", "Number of simultaneous connections", true},
+{"numIterations", "Number of iterations to run. Each query is sent to 
each connection this many times", true}
+};
+
+std::map qsOptionValues;
+
+const char* exceptionInject="alter session set 
`drill.exec.testing.controls` = '{ \"injections\" : [{ 
\"type\":\"exception\",\"siteClass\":\"org.apache.drill.exec.work.fragment.FragmentExecutor\",\"desc\":\"fragment-execution\",\"nSkip\":0,\"nFire\":1,\"exceptionClass\":\"java.lang.OutOfMemoryError\"}]}'";
--- End diff --

not valid; missing injection site in FragmentExecutor?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[sudheeshkatkam]

Github user sudheeshkatkam commented on a diff in the pull request:

https://github.com/apache/drill/pull/396#discussion_r55112202
  
--- Diff: contrib/native/client/example/pooledConnections.cpp ---
@@ -0,0 +1,300 @@
+/*
+ * 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.
+ */
+
+#include 
+#include 
+#include 
+#include 
+#include 
+#include "drill/drillc.hpp"
+
+int nOptions=5;
+
+struct Option{
+char name[32];
+char desc[128];
+bool required;
+}qsOptions[]= {
+{"query", "Query strings, separated by semicolons", true},
+{"connectStr", "Connect string", true},
+{"logLevel", "Logging level [trace|debug|info|warn|error|fatal]", 
false},
+{"numConnections", "Number of simultaneous connections", true},
+{"numIterations", "Number of iterations to run. Each query is sent to 
each connection this many times", true}
+};
+
+std::map qsOptionValues;
+
+const char* exceptionInject="alter session set 
`drill.exec.testing.controls` = '{ \"injections\" : [{ 
\"type\":\"exception\",\"siteClass\":\"org.apache.drill.exec.work.fragment.FragmentExecutor\",\"desc\":\"fragment-execution\",\"nSkip\":0,\"nFire\":1,\"exceptionClass\":\"java.lang.OutOfMemoryError\"}]}'";
+
+Drill::status_t SchemaListener(void* ctx, Drill::FieldDefPtr fields, 
Drill::DrillClientError* err){
+if(!err){
+printf("SCHEMA CHANGE DETECTED:\n");
+for(size_t i=0; isize(); i++){
+std::string name= fields->at(i)->getName();
+printf("%s\t", name.c_str());
+}
+printf("\n");
+return Drill::QRY_SUCCESS ;
+}else{
+std::cerr<< "ERROR: " << err->msg << std::endl;
+return Drill::QRY_FAILURE;
+}
+}
+
+boost::mutex listenerMutex;
+Drill::status_t QueryResultsListener(void* ctx, Drill::RecordBatch* b, 
Drill::DrillClientError* err){
+boost::lock_guard listenerLock(listenerMutex);
+if(!err){
+if(b!=NULL){
+b->print(std::cout, 0); // print all rows
+std::cout << "DATA RECEIVED ..." << std::endl;
+delete b; // we're done with this batch, we can delete it
+return Drill::QRY_FAILURE;
+}else{
+std::cout << "Query Complete." << std::endl;
+return Drill::QRY_SUCCESS;
+   }
+}else{
+assert(b==NULL);
+switch(err->status) {
+case Drill::QRY_COMPLETED:
+case Drill::QRY_CANCELED:
+std::cerr<< "INFO: " << err->msg << std::endl;
+return Drill::QRY_SUCCESS;
--- End diff --

Confusing, since query succeeds when error != null ?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[sudheeshkatkam]

Github user sudheeshkatkam commented on a diff in the pull request:

https://github.com/apache/drill/pull/396#discussion_r55112213
  
--- Diff: contrib/native/client/src/clientlib/drillClientImpl.cpp ---
@@ -1392,6 +1387,198 @@ void DrillClientQueryResult::clearAndDestroy(){
 }
 }
 
+
+connectionStatus_t PooledDrillClientImpl::connect(const char* connStr){
+connectionStatus_t stat = CONN_SUCCESS;
+std::string pathToDrill, protocol, hostPortStr;
+std::string host;
+std::string port;
+m_connectStr=connStr;
+Utils::parseConnectStr(connStr, pathToDrill, protocol, hostPortStr);
+if(!strcmp(protocol.c_str(), "zk")){
+// Get a list of drillbits
+ZookeeperImpl zook;
+std::vector drillbits;
+int err = zook.getAllDrillbits(hostPortStr.c_str(), 
pathToDrill.c_str(), drillbits);
+if(!err){
+Utils::shuffle(drillbits);
+// The original shuffled order is maintained if we shuffle 
first and then add any missing elements
+Utils::add(m_drillbits, drillbits);
+exec::DrillbitEndpoint e;
+size_t nextIndex=0;
+boost::lock_guard cLock(m_cMutex);
+m_lastConnection++;
+nextIndex = (m_lastConnection)%(getDrillbitCount());
+DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Pooled Connection"
+<< "(" << (void*)this << ")"
+<< ": Current counter is: " 
+<< m_lastConnection << std::endl;)
+err=zook.getEndPoint(m_drillbits, nextIndex, e);
+if(!err){
+host=boost::lexical_cast(e.address());
+port=boost::lexical_cast(e.user_port());
+}
+}
+if(err){
+return handleConnError(CONN_ZOOKEEPER_ERROR, 
getMessage(ERR_CONN_ZOOKEEPER, zook.getError().c_str()));
+}
+zook.close();
+m_bIsDirectConnection=false;
+}else if(!strcmp(protocol.c_str(), "local")){
+char tempStr[MAX_CONNECT_STR+1];
+strncpy(tempStr, hostPortStr.c_str(), MAX_CONNECT_STR); 
tempStr[MAX_CONNECT_STR]=0;
+host=strtok(tempStr, ":");
+port=strtok(NULL, "");
+m_bIsDirectConnection=true;
+}else{
+return handleConnError(CONN_INVALID_INPUT, 
getMessage(ERR_CONN_UNKPROTO, protocol.c_str()));
+}
+DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Connecting to endpoint: (Pooled) 
" << host << ":" << port << std::endl;)
+DrillClientImpl* pDrillClientImpl = new DrillClientImpl();
+stat =  pDrillClientImpl->connect(host.c_str(), port.c_str());
+if(stat == CONN_SUCCESS){
+m_clientConnections.push_back(pDrillClientImpl);
+}else{
+DrillClientError* pErr = pDrillClientImpl->getError();
+handleConnError((connectionStatus_t)pErr->status, pErr->msg);
+delete pDrillClientImpl;
+}
+return stat;
+}
+
+connectionStatus_t 
PooledDrillClientImpl::validateHandshake(DrillUserProperties* props){
+// Assume there is one valid connection to at least one drillbit
+connectionStatus_t stat=CONN_FAILURE;
+// Keep a copy of the user properties
+if(props!=NULL){
+m_pUserProperties = new DrillUserProperties;
+for(size_t i=0; isize(); i++){
+m_pUserProperties->setProperty(
+props->keyAt(i),
+props->valueAt(i)
+);
+}
+}
+DrillClientImpl* pDrillClientImpl = getOneConnection();
+if(pDrillClientImpl != NULL){
+DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Validating handshake: 
(Pooled) " << pDrillClientImpl->m_connectedHost << std::endl;)
+stat=pDrillClientImpl->validateHandshake(m_pUserProperties);
+}
+else{
+stat =  handleConnError(CONN_NOTCONNECTED, 
getMessage(ERR_CONN_NOCONN));
+}
+return stat;
+}
+
+DrillClientQueryResult* 
PooledDrillClientImpl::SubmitQuery(::exec::shared::QueryType t, const 
std::string& plan, pfnQueryResultsListener listener, void* listenerCtx){
+DrillClientQueryResult* pDrillClientQueryResult = NULL;
+DrillClientImpl* pDrillClientImpl = NULL;
+pDrillClientImpl = getOneConnection();
+if(pDrillClientImpl != NULL){
+
pDrillClientQueryResult=pDrillClientImpl->SubmitQuery(t,plan,listener,listenerCtx);
+m_queriesExecuted++;
+}
+return pDrillClientQueryResult;
+}
+
+void PooledDrillClientImpl::freeQueryResources(DrillClientQueryResult* 
pQryResult){
+// Nothing to do. If this class ever keeps track of executing queries 
then it will need 
+// 

[GitHub] drill pull request: DRILL-4313: Improve method of picking a random...

[parthchandra]

GitHub user parthchandra opened a pull request:

https://github.com/apache/drill/pull/396

DRILL-4313: Improve method of picking a random drillbit from the C++ client

Submitting as multiple commits to make review easier
Changes -
1) Made the logging thread safe. 
2) Use Boost random method generation instead of C rand() method.
3) Include git properties in build and print build commit id in log.
4) Add interface to get client error based on query handle. (Previous API 
always got back the last error that occurred). This could lead to clobbering of 
error messages when multiple errors were encountered.
5) Allow pooled client connections. Every query submitted used a different 
connection from the pool. The pool is filled lazily. Disabled by default . Can 
be enabled by (and pool size defined by) setting environment variable(s).


You can merge this pull request into a Git repository by running:

$ git pull https://github.com/parthchandra/incubator-drill DRILL-4313_2

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/drill/pull/396.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #396


commit a03ae1ec30ed28a1e15ffc8162ab560fdd544411
Author: Parth Chandra 
Date:   2016-02-12T23:42:53Z

DRILL-4313: C++ Client - Thread safe Logging

commit b494dc184578853167a3f6330c92beb475b74835
Author: Parth Chandra 
Date:   2016-02-12T22:03:37Z

DRILL-4313: Update random drill bit selection. Shuffle the list initially, 
then round robin.
Add Utility methods to get random numbers and to shuffle and add vectors.
Whitespace cleanup

commit 629142a295ecb4c79f37f31d5dd34d4b8a889638
Author: Parth Chandra 
Date:   2016-02-18T23:53:35Z

DRILL-4313: C++ client - Add Git properties to build and print to log.

commit 273cfc4bac96e4b65f66e6a7eea84642d50ecf50
Author: Parth Chandra 
Date:   2016-02-22T19:21:58Z

DRILL-4313: - Add interface to get error based on query handle.

commit f399792c2ca48e5d2c41b3daba4a2a3274992fcc
Author: Parth Chandra 
Date:   2016-02-13T06:38:51Z

DRILL-4313: Add support for Pooled connections.
Allows switching between pooled and unpooled connections based on
environment variables

Conflicts:
contrib/native/client/src/clientlib/utils.hpp




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---