16 package com.mapd.logrunner;
18 import org.apache.thrift.TException;
19 import org.apache.thrift.protocol.TJSONProtocol;
20 import org.apache.thrift.protocol.TProtocol;
21 import org.apache.thrift.transport.THttpClient;
22 import org.apache.thrift.transport.TTransport;
23 import org.apache.thrift.transport.TTransportException;
24 import org.slf4j.Logger;
25 import org.slf4j.LoggerFactory;
27 import java.io.BufferedReader;
28 import java.io.FileNotFoundException;
29 import java.io.FileOutputStream;
30 import java.io.FileReader;
31 import java.io.IOException;
32 import java.util.ArrayList;
33 import java.util.HashMap;
34 import java.util.List;
36 import java.util.concurrent.ArrayBlockingQueue;
37 import java.util.concurrent.ExecutorService;
38 import java.util.concurrent.Executors;
39 import java.util.concurrent.ThreadPoolExecutor;
40 import java.util.concurrent.TimeUnit;
41 import java.util.logging.Level;
43 import ai.heavy.thrift.server.Heavy;
44 import ai.heavy.thrift.server.TColumn;
45 import ai.heavy.thrift.server.TColumnData;
46 import ai.heavy.thrift.server.TColumnType;
47 import ai.heavy.thrift.server.TDBException;
48 import ai.heavy.thrift.server.TDBInfo;
49 import ai.heavy.thrift.server.TDatum;
50 import ai.heavy.thrift.server.TExecuteMode;
51 import ai.heavy.thrift.server.TPixel;
52 import ai.heavy.thrift.server.TQueryResult;
53 import ai.heavy.thrift.server.TRenderResult;
54 import ai.heavy.thrift.server.TRow;
55 import ai.heavy.thrift.server.TRowSet;
56 import ai.heavy.thrift.server.TTableDetails;
63 final static Logger
logger = LoggerFactory.getLogger(LogRunner.class);
66 private HashMap<Integer, String>
json;
70 public static void main(String[]
args)
throws TException {
71 logger.info(
"Hello, World");
76 }
catch (TTransportException ex) {
77 logger.error(ex.toString());
83 sqlquery =
new HashMap<Integer, String>();
85 json =
new HashMap<Integer, String>();
88 void doWork(String[]
args)
throws TTransportException, TException {
89 logger.info(
"In doWork here");
91 int numberThreads = 3;
101 logger.info(
"got session");
104 ExecutorService executor =
new ThreadPoolExecutor(numberThreads,
107 TimeUnit.MILLISECONDS,
108 new ArrayBlockingQueue<Runnable>(15),
109 new ThreadPoolExecutor.CallerRunsPolicy());
113 BufferedReader in =
new BufferedReader(
new FileReader(
args[2]));
116 while ((str = in.readLine()) != null) {
117 Runnable worker =
new myThread(str, client, session);
122 logger.info(
"############loop complete");
125 }
catch (IOException e) {
126 logger.error(
"IOException " + e.getMessage());
130 private Heavy.Client
getClient(String hostname,
int port)
throws TTransportException {
131 TTransport transport = null;
134 transport =
new THttpClient(
"http://" + hostname +
":" + port);
139 TProtocol protocol =
new TJSONProtocol(transport);
142 return new Heavy.Client(protocol);
146 throws TTransportException, TDBException, TException {
147 String session = client.connect(
"mapd",
"HyperInteractive",
"mapd");
148 logger.info(
"Connected session is " + session);
153 throws TDBException, TException {
155 logger.info(
"Trying to disconnect session " + session);
156 client.disconnect(session);
159 private void theRest(Heavy.Client client, String session)
throws TException {
161 List<TDBInfo> dbs = client.get_databases(session);
163 for (TDBInfo db : dbs) {
164 logger.info(
"db is " + db.toString());
168 List<String>
tables = client.get_tables(session);
170 for (String tab : tables) {
171 logger.info(
"Tables is " + tab);
175 logger.info(
"Version " + client.get_version());
178 TTableDetails table_details = client.get_table_details(session,
"flights");
179 for (TColumnType col : table_details.row_desc) {
180 logger.info(
"col name :" + col.col_name);
181 logger.info(
"\tcol encoding :" + col.col_type.encoding);
182 logger.info(
"\tcol is_array :" + col.col_type.is_array);
183 logger.info(
"\tcol nullable :" + col.col_type.nullable);
188 logger.info(
" -- before query -- ");
190 TQueryResult sql_execute = client.sql_execute(session,
191 "Select uniquecarrier,flightnum from flights LIMIT 3;",
200 logger.info(
" -- after query -- ");
202 logger.info(
"TQueryResult execution time is " + sql_execute.getExecution_time_ms());
203 logger.info(
"TQueryResult is " + sql_execute.toString());
204 logger.info(
"TQueryResult getFieldValue is "
205 + sql_execute.getFieldValue(TQueryResult._Fields.ROW_SET));
207 TRowSet row_set = sql_execute.getRow_set();
208 Object fieldValue = sql_execute.getFieldValue(TQueryResult._Fields.ROW_SET);
210 logger.info(
"fieldValue " + fieldValue);
212 logger.info(
"TRowSet is " + row_set.toString());
214 logger.info(
"Get rows size " + row_set.getRowsSize());
215 logger.info(
"Get col size " + row_set.getRowsSize());
217 List<TRow> rows = row_set.getRows();
219 for (TRow row : rows) {
220 List<TDatum> cols = row.getCols();
222 for (TDatum dat : cols) {
223 logger.info(
"ROW " + count +
" " + dat.getFieldValue(TDatum._Fields.VAL));
229 List<TColumn> columns = row_set.getColumns();
231 logger.info(
"columns " + columns);
233 for (TColumn col : columns) {
234 TColumnData data = col.getData();
236 logger.info(
"COL " + count +
" " + data.toString());
247 myThread(String str1, Heavy.Client client1, String session1) {
255 int logStart = str.indexOf(
']');
256 if (logStart != -1) {
257 String det = str.substring(logStart + 1).trim();
258 String header = str.substring(0, logStart).trim();
260 String[] headDet = header.split(
" .");
263 if (headDet.length != 4 || headDet[0].equals(
"Log")) {
266 Integer pid = Integer.valueOf(headDet[2]);
268 if (det.contains(
"sql_execute :")) {
269 logger.info(
"det " + det);
270 String sl[] = det.split(
":query_str:");
271 logger.info(
"run query " + sl[1]);
273 client.sql_execute(
session, sl[1],
true, null, -1, -1);
274 }
catch (TDBException ex1) {
276 "Failed to execute " + sl[1] +
" exception " + ex1.getError_msg());
277 }
catch (TException ex) {
278 logger.error(
"Failed to execute " + sl[1] +
" exception " + ex.toString());
287 if (det.contains(
"get_result_row_for_pixel :")) {
288 logger.info(
"det " + det);
289 String ss[] = det.split(
":");
290 String sl[] = det.split(
":table_col_names:");
291 logger.info(
"run get_result_for_pixel " + sl[1]);
292 Map<String, List<String>> tcn =
new HashMap<String, List<String>>();
294 String tn[] = sl[1].split(
":");
295 for (
int i = 0; i < tn.length; i++) {
296 String
name[] = tn[i].split(
",");
297 List<String> col =
new ArrayList<String>();
298 for (
int j = 1; j < name.length; j++) {
301 tcn.put(name[0], col);
304 client.get_result_row_for_pixel(
session,
305 Integer.parseInt(ss[3]),
306 new TPixel(
Integer.parseInt(ss[5]), Integer.parseInt(ss[7])),
309 Integer.parseInt(ss[11]),
311 }
catch (TDBException ex1) {
312 logger.error(
"Failed to execute get_result_row_for_pixel exception "
313 + ex1.getError_msg());
314 }
catch (TException ex) {
315 logger.error(
"Failed to execute get_result_row_for_pixel exception "
321 if (det.contains(
"render_vega :")) {
322 logger.info(
"det " + det);
323 String ss[] = det.split(
":");
324 String sl[] = det.split(
":vega_json:");
325 json.put(pid, det.substring(det.indexOf(
"render_vega :") + 13, det.length()));
326 logger.info(
"JSON = " + sl[1]);
327 logger.info(
"widget = " + Integer.parseInt(ss[3]));
328 logger.info(
"compressionLevel = " + Integer.parseInt(ss[5]));
329 logger.info(
"run render_vega");
331 logger.info(
"In render: setting gpu mode as we were in CPU mode");
335 client.set_execution_mode(
session, TExecuteMode.GPU);
336 }
catch (TException ex) {
337 logger.error(
"Failed to set_execution_mode exception " + ex.toString());
341 TRenderResult fred = client.render_vega(
session,
342 Integer.parseInt(ss[3]),
348 FileOutputStream fos;
350 fos =
new FileOutputStream(
"/tmp/png.png");
352 fred.image.position(0);
353 byte[] tgxImageDataByte =
new byte[fred.image.limit()];
354 fred.image.get(tgxImageDataByte);
355 fos.write(tgxImageDataByte);
357 }
catch (FileNotFoundException ex) {
358 logger.error(
"Failed to create file exception " + ex.toString());
359 }
catch (IOException ex) {
360 logger.error(
"Failed to create file exception " + ex.toString());
364 }
catch (TException ex) {
365 logger.error(
"Failed to execute render_vega exception " + ex.toString());
370 if (det.contains(
"User mapd sets CPU mode")) {
371 logger.info(
"Set cpu mode");
375 client.set_execution_mode(
session, TExecuteMode.CPU);
376 }
catch (TException ex) {
377 logger.error(
"Failed to set_execution_mode exception " + ex.toString());
382 if (det.contains(
"User mapd sets GPU mode")) {
383 logger.info(
"Set gpu mode");
387 client.set_execution_mode(
session, TExecuteMode.GPU);
388 }
catch (TException ex) {
390 "Failed to execute set_execution_mode exception " + ex.toString());
myThread(String str1, Heavy.Client client1, String session1)
HashMap< Integer, String > sqlquery
void doWork(String[] args)
void theRest(Heavy.Client client, String session)
HashMap< Integer, String > originalSql
static final Logger logger
HashMap< Integer, String > json
String getSession(Heavy.Client client)
static void main(String[] args)
void closeSession(Heavy.Client client, String session)
Heavy.Client getClient(String hostname, int port)