Selaa lähdekoodia

HUE-639. thrift enhancements for sync query execution and additional error diagnostics

Prasad Mujumdar 13 vuotta sitten
vanhempi
commit
cafbb72e1d

+ 25 - 4
apps/beeswax/gen-py/beeswaxd/BeeswaxService-remote

@@ -22,14 +22,17 @@ if len(sys.argv) <= 1 or sys.argv[1] == '--help':
   print ''
   print 'Functions:'
   print '  QueryHandle query(Query query)'
+  print '  QueryHandle executeAndWait(Query query, LogContextId clientCtx)'
   print '  QueryExplanation explain(Query query)'
-  print '  Results fetch(QueryHandle query_id, bool start_over)'
+  print '  Results fetch(QueryHandle query_id, bool start_over, i32 fetch_size)'
   print '  QueryState get_state(QueryHandle handle)'
   print '  ResultsMetadata get_results_metadata(QueryHandle handle)'
   print '  string echo(string s)'
   print '  string dump_config()'
   print '  string get_log(LogContextId context)'
   print '   get_default_configuration(bool include_hadoop)'
+  print '  void close(QueryHandle handle)'
+  print '  void clean(LogContextId log_context)'
   print ''
   sys.exit(0)
 
@@ -86,6 +89,12 @@ if cmd == 'query':
     sys.exit(1)
   pp.pprint(client.query(eval(args[0]),))
 
+elif cmd == 'executeAndWait':
+  if len(args) != 2:
+    print 'executeAndWait requires 2 args'
+    sys.exit(1)
+  pp.pprint(client.executeAndWait(eval(args[0]),eval(args[1]),))
+
 elif cmd == 'explain':
   if len(args) != 1:
     print 'explain requires 1 args'
@@ -93,10 +102,10 @@ elif cmd == 'explain':
   pp.pprint(client.explain(eval(args[0]),))
 
 elif cmd == 'fetch':
-  if len(args) != 2:
-    print 'fetch requires 2 args'
+  if len(args) != 3:
+    print 'fetch requires 3 args'
     sys.exit(1)
-  pp.pprint(client.fetch(eval(args[0]),eval(args[1]),))
+  pp.pprint(client.fetch(eval(args[0]),eval(args[1]),eval(args[2]),))
 
 elif cmd == 'get_state':
   if len(args) != 1:
@@ -134,6 +143,18 @@ elif cmd == 'get_default_configuration':
     sys.exit(1)
   pp.pprint(client.get_default_configuration(eval(args[0]),))
 
+elif cmd == 'close':
+  if len(args) != 1:
+    print 'close requires 1 args'
+    sys.exit(1)
+  pp.pprint(client.close(eval(args[0]),))
+
+elif cmd == 'clean':
+  if len(args) != 1:
+    print 'clean requires 1 args'
+    sys.exit(1)
+  pp.pprint(client.clean(eval(args[0]),))
+
 else:
   print 'Unrecognized method %s' % cmd
   sys.exit(1)

+ 572 - 8
apps/beeswax/gen-py/beeswaxd/BeeswaxService.py

@@ -25,6 +25,16 @@ class Iface(object):
     """
     pass
 
+  def executeAndWait(self, query, clientCtx):
+    """
+    run a query synchronously and return a handle (QueryHandle).
+
+    Parameters:
+     - query
+     - clientCtx
+    """
+    pass
+
   def explain(self, query):
     """
     Get the query plan for a query.
@@ -34,14 +44,16 @@ class Iface(object):
     """
     pass
 
-  def fetch(self, query_id, start_over):
+  def fetch(self, query_id, start_over, fetch_size):
     """
     Get the results of a query. This is non-blocking. Caller should check
-    Results.ready to determine if the results are in yet.
+    Results.ready to determine if the results are in yet. The call requests
+    the batch size of fetch.
 
     Parameters:
      - query_id
      - start_over
+     - fetch_size
     """
     pass
 
@@ -95,6 +107,20 @@ class Iface(object):
     """
     pass
 
+  def close(self, handle):
+    """
+    Parameters:
+     - handle
+    """
+    pass
+
+  def clean(self, log_context):
+    """
+    Parameters:
+     - log_context
+    """
+    pass
+
 
 class Client(Iface):
   def __init__(self, iprot, oprot=None):
@@ -137,6 +163,42 @@ class Client(Iface):
       raise result.error
     raise TApplicationException(TApplicationException.MISSING_RESULT, "query failed: unknown result");
 
+  def executeAndWait(self, query, clientCtx):
+    """
+    run a query synchronously and return a handle (QueryHandle).
+
+    Parameters:
+     - query
+     - clientCtx
+    """
+    self.send_executeAndWait(query, clientCtx)
+    return self.recv_executeAndWait()
+
+  def send_executeAndWait(self, query, clientCtx):
+    self._oprot.writeMessageBegin('executeAndWait', TMessageType.CALL, self._seqid)
+    args = executeAndWait_args()
+    args.query = query
+    args.clientCtx = clientCtx
+    args.write(self._oprot)
+    self._oprot.writeMessageEnd()
+    self._oprot.trans.flush()
+
+  def recv_executeAndWait(self, ):
+    (fname, mtype, rseqid) = self._iprot.readMessageBegin()
+    if mtype == TMessageType.EXCEPTION:
+      x = TApplicationException()
+      x.read(self._iprot)
+      self._iprot.readMessageEnd()
+      raise x
+    result = executeAndWait_result()
+    result.read(self._iprot)
+    self._iprot.readMessageEnd()
+    if result.success is not None:
+      return result.success
+    if result.error is not None:
+      raise result.error
+    raise TApplicationException(TApplicationException.MISSING_RESULT, "executeAndWait failed: unknown result");
+
   def explain(self, query):
     """
     Get the query plan for a query.
@@ -171,23 +233,26 @@ class Client(Iface):
       raise result.error
     raise TApplicationException(TApplicationException.MISSING_RESULT, "explain failed: unknown result");
 
-  def fetch(self, query_id, start_over):
+  def fetch(self, query_id, start_over, fetch_size):
     """
     Get the results of a query. This is non-blocking. Caller should check
-    Results.ready to determine if the results are in yet.
+    Results.ready to determine if the results are in yet. The call requests
+    the batch size of fetch.
 
     Parameters:
      - query_id
      - start_over
+     - fetch_size
     """
-    self.send_fetch(query_id, start_over)
+    self.send_fetch(query_id, start_over, fetch_size)
     return self.recv_fetch()
 
-  def send_fetch(self, query_id, start_over):
+  def send_fetch(self, query_id, start_over, fetch_size):
     self._oprot.writeMessageBegin('fetch', TMessageType.CALL, self._seqid)
     args = fetch_args()
     args.query_id = query_id
     args.start_over = start_over
+    args.fetch_size = fetch_size
     args.write(self._oprot)
     self._oprot.writeMessageEnd()
     self._oprot.trans.flush()
@@ -403,12 +468,73 @@ class Client(Iface):
       return result.success
     raise TApplicationException(TApplicationException.MISSING_RESULT, "get_default_configuration failed: unknown result");
 
+  def close(self, handle):
+    """
+    Parameters:
+     - handle
+    """
+    self.send_close(handle)
+    self.recv_close()
+
+  def send_close(self, handle):
+    self._oprot.writeMessageBegin('close', TMessageType.CALL, self._seqid)
+    args = close_args()
+    args.handle = handle
+    args.write(self._oprot)
+    self._oprot.writeMessageEnd()
+    self._oprot.trans.flush()
+
+  def recv_close(self, ):
+    (fname, mtype, rseqid) = self._iprot.readMessageBegin()
+    if mtype == TMessageType.EXCEPTION:
+      x = TApplicationException()
+      x.read(self._iprot)
+      self._iprot.readMessageEnd()
+      raise x
+    result = close_result()
+    result.read(self._iprot)
+    self._iprot.readMessageEnd()
+    if result.error is not None:
+      raise result.error
+    if result.error2 is not None:
+      raise result.error2
+    return
+
+  def clean(self, log_context):
+    """
+    Parameters:
+     - log_context
+    """
+    self.send_clean(log_context)
+    self.recv_clean()
+
+  def send_clean(self, log_context):
+    self._oprot.writeMessageBegin('clean', TMessageType.CALL, self._seqid)
+    args = clean_args()
+    args.log_context = log_context
+    args.write(self._oprot)
+    self._oprot.writeMessageEnd()
+    self._oprot.trans.flush()
+
+  def recv_clean(self, ):
+    (fname, mtype, rseqid) = self._iprot.readMessageBegin()
+    if mtype == TMessageType.EXCEPTION:
+      x = TApplicationException()
+      x.read(self._iprot)
+      self._iprot.readMessageEnd()
+      raise x
+    result = clean_result()
+    result.read(self._iprot)
+    self._iprot.readMessageEnd()
+    return
+
 
 class Processor(Iface, TProcessor):
   def __init__(self, handler):
     self._handler = handler
     self._processMap = {}
     self._processMap["query"] = Processor.process_query
+    self._processMap["executeAndWait"] = Processor.process_executeAndWait
     self._processMap["explain"] = Processor.process_explain
     self._processMap["fetch"] = Processor.process_fetch
     self._processMap["get_state"] = Processor.process_get_state
@@ -417,6 +543,8 @@ class Processor(Iface, TProcessor):
     self._processMap["dump_config"] = Processor.process_dump_config
     self._processMap["get_log"] = Processor.process_get_log
     self._processMap["get_default_configuration"] = Processor.process_get_default_configuration
+    self._processMap["close"] = Processor.process_close
+    self._processMap["clean"] = Processor.process_clean
 
   def process(self, iprot, oprot):
     (name, type, seqid) = iprot.readMessageBegin()
@@ -447,6 +575,20 @@ class Processor(Iface, TProcessor):
     oprot.writeMessageEnd()
     oprot.trans.flush()
 
+  def process_executeAndWait(self, seqid, iprot, oprot):
+    args = executeAndWait_args()
+    args.read(iprot)
+    iprot.readMessageEnd()
+    result = executeAndWait_result()
+    try:
+      result.success = self._handler.executeAndWait(args.query, args.clientCtx)
+    except BeeswaxException, error:
+      result.error = error
+    oprot.writeMessageBegin("executeAndWait", TMessageType.REPLY, seqid)
+    result.write(oprot)
+    oprot.writeMessageEnd()
+    oprot.trans.flush()
+
   def process_explain(self, seqid, iprot, oprot):
     args = explain_args()
     args.read(iprot)
@@ -467,7 +609,7 @@ class Processor(Iface, TProcessor):
     iprot.readMessageEnd()
     result = fetch_result()
     try:
-      result.success = self._handler.fetch(args.query_id, args.start_over)
+      result.success = self._handler.fetch(args.query_id, args.start_over, args.fetch_size)
     except QueryNotFoundException, error:
       result.error = error
     except BeeswaxException, error2:
@@ -552,6 +694,33 @@ class Processor(Iface, TProcessor):
     oprot.writeMessageEnd()
     oprot.trans.flush()
 
+  def process_close(self, seqid, iprot, oprot):
+    args = close_args()
+    args.read(iprot)
+    iprot.readMessageEnd()
+    result = close_result()
+    try:
+      self._handler.close(args.handle)
+    except QueryNotFoundException, error:
+      result.error = error
+    except BeeswaxException, error2:
+      result.error2 = error2
+    oprot.writeMessageBegin("close", TMessageType.REPLY, seqid)
+    result.write(oprot)
+    oprot.writeMessageEnd()
+    oprot.trans.flush()
+
+  def process_clean(self, seqid, iprot, oprot):
+    args = clean_args()
+    args.read(iprot)
+    iprot.readMessageEnd()
+    result = clean_result()
+    self._handler.clean(args.log_context)
+    oprot.writeMessageBegin("clean", TMessageType.REPLY, seqid)
+    result.write(oprot)
+    oprot.writeMessageEnd()
+    oprot.trans.flush()
+
 
 # HELPER FUNCTIONS AND STRUCTURES
 
@@ -689,6 +858,152 @@ class query_result(object):
   def __ne__(self, other):
     return not (self == other)
 
+class executeAndWait_args(object):
+  """
+  Attributes:
+   - query
+   - clientCtx
+  """
+
+  thrift_spec = (
+    None, # 0
+    (1, TType.STRUCT, 'query', (Query, Query.thrift_spec), None, ), # 1
+    (2, TType.STRING, 'clientCtx', None, None, ), # 2
+  )
+
+  def __init__(self, query=None, clientCtx=None,):
+    self.query = query
+    self.clientCtx = clientCtx
+
+  def read(self, iprot):
+    if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
+      fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec))
+      return
+    iprot.readStructBegin()
+    while True:
+      (fname, ftype, fid) = iprot.readFieldBegin()
+      if ftype == TType.STOP:
+        break
+      if fid == 1:
+        if ftype == TType.STRUCT:
+          self.query = Query()
+          self.query.read(iprot)
+        else:
+          iprot.skip(ftype)
+      elif fid == 2:
+        if ftype == TType.STRING:
+          self.clientCtx = iprot.readString();
+        else:
+          iprot.skip(ftype)
+      else:
+        iprot.skip(ftype)
+      iprot.readFieldEnd()
+    iprot.readStructEnd()
+
+  def write(self, oprot):
+    if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None:
+      oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec)))
+      return
+    oprot.writeStructBegin('executeAndWait_args')
+    if self.query is not None:
+      oprot.writeFieldBegin('query', TType.STRUCT, 1)
+      self.query.write(oprot)
+      oprot.writeFieldEnd()
+    if self.clientCtx is not None:
+      oprot.writeFieldBegin('clientCtx', TType.STRING, 2)
+      oprot.writeString(self.clientCtx)
+      oprot.writeFieldEnd()
+    oprot.writeFieldStop()
+    oprot.writeStructEnd()
+
+  def validate(self):
+    return
+
+
+  def __repr__(self):
+    L = ['%s=%r' % (key, value)
+      for key, value in self.__dict__.iteritems()]
+    return '%s(%s)' % (self.__class__.__name__, ', '.join(L))
+
+  def __eq__(self, other):
+    return isinstance(other, self.__class__) and self.__dict__ == other.__dict__
+
+  def __ne__(self, other):
+    return not (self == other)
+
+class executeAndWait_result(object):
+  """
+  Attributes:
+   - success
+   - error
+  """
+
+  thrift_spec = (
+    (0, TType.STRUCT, 'success', (QueryHandle, QueryHandle.thrift_spec), None, ), # 0
+    (1, TType.STRUCT, 'error', (BeeswaxException, BeeswaxException.thrift_spec), None, ), # 1
+  )
+
+  def __init__(self, success=None, error=None,):
+    self.success = success
+    self.error = error
+
+  def read(self, iprot):
+    if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
+      fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec))
+      return
+    iprot.readStructBegin()
+    while True:
+      (fname, ftype, fid) = iprot.readFieldBegin()
+      if ftype == TType.STOP:
+        break
+      if fid == 0:
+        if ftype == TType.STRUCT:
+          self.success = QueryHandle()
+          self.success.read(iprot)
+        else:
+          iprot.skip(ftype)
+      elif fid == 1:
+        if ftype == TType.STRUCT:
+          self.error = BeeswaxException()
+          self.error.read(iprot)
+        else:
+          iprot.skip(ftype)
+      else:
+        iprot.skip(ftype)
+      iprot.readFieldEnd()
+    iprot.readStructEnd()
+
+  def write(self, oprot):
+    if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None:
+      oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec)))
+      return
+    oprot.writeStructBegin('executeAndWait_result')
+    if self.success is not None:
+      oprot.writeFieldBegin('success', TType.STRUCT, 0)
+      self.success.write(oprot)
+      oprot.writeFieldEnd()
+    if self.error is not None:
+      oprot.writeFieldBegin('error', TType.STRUCT, 1)
+      self.error.write(oprot)
+      oprot.writeFieldEnd()
+    oprot.writeFieldStop()
+    oprot.writeStructEnd()
+
+  def validate(self):
+    return
+
+
+  def __repr__(self):
+    L = ['%s=%r' % (key, value)
+      for key, value in self.__dict__.iteritems()]
+    return '%s(%s)' % (self.__class__.__name__, ', '.join(L))
+
+  def __eq__(self, other):
+    return isinstance(other, self.__class__) and self.__dict__ == other.__dict__
+
+  def __ne__(self, other):
+    return not (self == other)
+
 class explain_args(object):
   """
   Attributes:
@@ -828,17 +1143,20 @@ class fetch_args(object):
   Attributes:
    - query_id
    - start_over
+   - fetch_size
   """
 
   thrift_spec = (
     None, # 0
     (1, TType.STRUCT, 'query_id', (QueryHandle, QueryHandle.thrift_spec), None, ), # 1
     (2, TType.BOOL, 'start_over', None, None, ), # 2
+    (3, TType.I32, 'fetch_size', None, -1, ), # 3
   )
 
-  def __init__(self, query_id=None, start_over=None,):
+  def __init__(self, query_id=None, start_over=None, fetch_size=thrift_spec[3][4],):
     self.query_id = query_id
     self.start_over = start_over
+    self.fetch_size = fetch_size
 
   def read(self, iprot):
     if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
@@ -860,6 +1178,11 @@ class fetch_args(object):
           self.start_over = iprot.readBool();
         else:
           iprot.skip(ftype)
+      elif fid == 3:
+        if ftype == TType.I32:
+          self.fetch_size = iprot.readI32();
+        else:
+          iprot.skip(ftype)
       else:
         iprot.skip(ftype)
       iprot.readFieldEnd()
@@ -878,6 +1201,10 @@ class fetch_args(object):
       oprot.writeFieldBegin('start_over', TType.BOOL, 2)
       oprot.writeBool(self.start_over)
       oprot.writeFieldEnd()
+    if self.fetch_size is not None:
+      oprot.writeFieldBegin('fetch_size', TType.I32, 3)
+      oprot.writeI32(self.fetch_size)
+      oprot.writeFieldEnd()
     oprot.writeFieldStop()
     oprot.writeStructEnd()
 
@@ -1718,6 +2045,243 @@ class get_default_configuration_result(object):
     return
 
 
+  def __repr__(self):
+    L = ['%s=%r' % (key, value)
+      for key, value in self.__dict__.iteritems()]
+    return '%s(%s)' % (self.__class__.__name__, ', '.join(L))
+
+  def __eq__(self, other):
+    return isinstance(other, self.__class__) and self.__dict__ == other.__dict__
+
+  def __ne__(self, other):
+    return not (self == other)
+
+class close_args(object):
+  """
+  Attributes:
+   - handle
+  """
+
+  thrift_spec = (
+    None, # 0
+    (1, TType.STRUCT, 'handle', (QueryHandle, QueryHandle.thrift_spec), None, ), # 1
+  )
+
+  def __init__(self, handle=None,):
+    self.handle = handle
+
+  def read(self, iprot):
+    if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
+      fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec))
+      return
+    iprot.readStructBegin()
+    while True:
+      (fname, ftype, fid) = iprot.readFieldBegin()
+      if ftype == TType.STOP:
+        break
+      if fid == 1:
+        if ftype == TType.STRUCT:
+          self.handle = QueryHandle()
+          self.handle.read(iprot)
+        else:
+          iprot.skip(ftype)
+      else:
+        iprot.skip(ftype)
+      iprot.readFieldEnd()
+    iprot.readStructEnd()
+
+  def write(self, oprot):
+    if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None:
+      oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec)))
+      return
+    oprot.writeStructBegin('close_args')
+    if self.handle is not None:
+      oprot.writeFieldBegin('handle', TType.STRUCT, 1)
+      self.handle.write(oprot)
+      oprot.writeFieldEnd()
+    oprot.writeFieldStop()
+    oprot.writeStructEnd()
+
+  def validate(self):
+    return
+
+
+  def __repr__(self):
+    L = ['%s=%r' % (key, value)
+      for key, value in self.__dict__.iteritems()]
+    return '%s(%s)' % (self.__class__.__name__, ', '.join(L))
+
+  def __eq__(self, other):
+    return isinstance(other, self.__class__) and self.__dict__ == other.__dict__
+
+  def __ne__(self, other):
+    return not (self == other)
+
+class close_result(object):
+  """
+  Attributes:
+   - error
+   - error2
+  """
+
+  thrift_spec = (
+    None, # 0
+    (1, TType.STRUCT, 'error', (QueryNotFoundException, QueryNotFoundException.thrift_spec), None, ), # 1
+    (2, TType.STRUCT, 'error2', (BeeswaxException, BeeswaxException.thrift_spec), None, ), # 2
+  )
+
+  def __init__(self, error=None, error2=None,):
+    self.error = error
+    self.error2 = error2
+
+  def read(self, iprot):
+    if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
+      fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec))
+      return
+    iprot.readStructBegin()
+    while True:
+      (fname, ftype, fid) = iprot.readFieldBegin()
+      if ftype == TType.STOP:
+        break
+      if fid == 1:
+        if ftype == TType.STRUCT:
+          self.error = QueryNotFoundException()
+          self.error.read(iprot)
+        else:
+          iprot.skip(ftype)
+      elif fid == 2:
+        if ftype == TType.STRUCT:
+          self.error2 = BeeswaxException()
+          self.error2.read(iprot)
+        else:
+          iprot.skip(ftype)
+      else:
+        iprot.skip(ftype)
+      iprot.readFieldEnd()
+    iprot.readStructEnd()
+
+  def write(self, oprot):
+    if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None:
+      oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec)))
+      return
+    oprot.writeStructBegin('close_result')
+    if self.error is not None:
+      oprot.writeFieldBegin('error', TType.STRUCT, 1)
+      self.error.write(oprot)
+      oprot.writeFieldEnd()
+    if self.error2 is not None:
+      oprot.writeFieldBegin('error2', TType.STRUCT, 2)
+      self.error2.write(oprot)
+      oprot.writeFieldEnd()
+    oprot.writeFieldStop()
+    oprot.writeStructEnd()
+
+  def validate(self):
+    return
+
+
+  def __repr__(self):
+    L = ['%s=%r' % (key, value)
+      for key, value in self.__dict__.iteritems()]
+    return '%s(%s)' % (self.__class__.__name__, ', '.join(L))
+
+  def __eq__(self, other):
+    return isinstance(other, self.__class__) and self.__dict__ == other.__dict__
+
+  def __ne__(self, other):
+    return not (self == other)
+
+class clean_args(object):
+  """
+  Attributes:
+   - log_context
+  """
+
+  thrift_spec = (
+    None, # 0
+    (1, TType.STRING, 'log_context', None, None, ), # 1
+  )
+
+  def __init__(self, log_context=None,):
+    self.log_context = log_context
+
+  def read(self, iprot):
+    if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
+      fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec))
+      return
+    iprot.readStructBegin()
+    while True:
+      (fname, ftype, fid) = iprot.readFieldBegin()
+      if ftype == TType.STOP:
+        break
+      if fid == 1:
+        if ftype == TType.STRING:
+          self.log_context = iprot.readString();
+        else:
+          iprot.skip(ftype)
+      else:
+        iprot.skip(ftype)
+      iprot.readFieldEnd()
+    iprot.readStructEnd()
+
+  def write(self, oprot):
+    if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None:
+      oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec)))
+      return
+    oprot.writeStructBegin('clean_args')
+    if self.log_context is not None:
+      oprot.writeFieldBegin('log_context', TType.STRING, 1)
+      oprot.writeString(self.log_context)
+      oprot.writeFieldEnd()
+    oprot.writeFieldStop()
+    oprot.writeStructEnd()
+
+  def validate(self):
+    return
+
+
+  def __repr__(self):
+    L = ['%s=%r' % (key, value)
+      for key, value in self.__dict__.iteritems()]
+    return '%s(%s)' % (self.__class__.__name__, ', '.join(L))
+
+  def __eq__(self, other):
+    return isinstance(other, self.__class__) and self.__dict__ == other.__dict__
+
+  def __ne__(self, other):
+    return not (self == other)
+
+class clean_result(object):
+
+  thrift_spec = (
+  )
+
+  def read(self, iprot):
+    if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
+      fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec))
+      return
+    iprot.readStructBegin()
+    while True:
+      (fname, ftype, fid) = iprot.readFieldBegin()
+      if ftype == TType.STOP:
+        break
+      else:
+        iprot.skip(ftype)
+      iprot.readFieldEnd()
+    iprot.readStructEnd()
+
+  def write(self, oprot):
+    if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None:
+      oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec)))
+      return
+    oprot.writeStructBegin('clean_result')
+    oprot.writeFieldStop()
+    oprot.writeStructEnd()
+
+  def validate(self):
+    return
+
+
   def __repr__(self):
     L = ['%s=%r' % (key, value)
       for key, value in self.__dict__.iteritems()]

+ 25 - 1
apps/beeswax/gen-py/beeswaxd/ttypes.py

@@ -498,6 +498,8 @@ class BeeswaxException(Exception):
    - message
    - log_context
    - handle
+   - errorCode
+   - SQLState
   """
 
   thrift_spec = (
@@ -505,12 +507,16 @@ class BeeswaxException(Exception):
     (1, TType.STRING, 'message', None, None, ), # 1
     (2, TType.STRING, 'log_context', None, None, ), # 2
     (3, TType.STRUCT, 'handle', (QueryHandle, QueryHandle.thrift_spec), None, ), # 3
+    (4, TType.I32, 'errorCode', None, 0, ), # 4
+    (5, TType.STRING, 'SQLState', None, "     ", ), # 5
   )
 
-  def __init__(self, message=None, log_context=None, handle=None,):
+  def __init__(self, message=None, log_context=None, handle=None, errorCode=thrift_spec[4][4], SQLState=thrift_spec[5][4],):
     self.message = message
     self.log_context = log_context
     self.handle = handle
+    self.errorCode = errorCode
+    self.SQLState = SQLState
 
   def read(self, iprot):
     if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
@@ -537,6 +543,16 @@ class BeeswaxException(Exception):
           self.handle.read(iprot)
         else:
           iprot.skip(ftype)
+      elif fid == 4:
+        if ftype == TType.I32:
+          self.errorCode = iprot.readI32();
+        else:
+          iprot.skip(ftype)
+      elif fid == 5:
+        if ftype == TType.STRING:
+          self.SQLState = iprot.readString();
+        else:
+          iprot.skip(ftype)
       else:
         iprot.skip(ftype)
       iprot.readFieldEnd()
@@ -559,6 +575,14 @@ class BeeswaxException(Exception):
       oprot.writeFieldBegin('handle', TType.STRUCT, 3)
       self.handle.write(oprot)
       oprot.writeFieldEnd()
+    if self.errorCode is not None:
+      oprot.writeFieldBegin('errorCode', TType.I32, 4)
+      oprot.writeI32(self.errorCode)
+      oprot.writeFieldEnd()
+    if self.SQLState is not None:
+      oprot.writeFieldBegin('SQLState', TType.STRING, 5)
+      oprot.writeString(self.SQLState)
+      oprot.writeFieldEnd()
     oprot.writeFieldStop()
     oprot.writeStructEnd()
 

+ 197 - 1
apps/beeswax/java/src/main/gen-java/com/cloudera/beeswax/api/BeeswaxException.java

@@ -27,16 +27,22 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
   private static final org.apache.thrift.protocol.TField MESSAGE_FIELD_DESC = new org.apache.thrift.protocol.TField("message", org.apache.thrift.protocol.TType.STRING, (short)1);
   private static final org.apache.thrift.protocol.TField LOG_CONTEXT_FIELD_DESC = new org.apache.thrift.protocol.TField("log_context", org.apache.thrift.protocol.TType.STRING, (short)2);
   private static final org.apache.thrift.protocol.TField HANDLE_FIELD_DESC = new org.apache.thrift.protocol.TField("handle", org.apache.thrift.protocol.TType.STRUCT, (short)3);
+  private static final org.apache.thrift.protocol.TField ERROR_CODE_FIELD_DESC = new org.apache.thrift.protocol.TField("errorCode", org.apache.thrift.protocol.TType.I32, (short)4);
+  private static final org.apache.thrift.protocol.TField SQLSTATE_FIELD_DESC = new org.apache.thrift.protocol.TField("SQLState", org.apache.thrift.protocol.TType.STRING, (short)5);
 
   public String message; // required
   public String log_context; // required
   public QueryHandle handle; // required
+  public int errorCode; // required
+  public String SQLState; // required
 
   /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
   public enum _Fields implements org.apache.thrift.TFieldIdEnum {
     MESSAGE((short)1, "message"),
     LOG_CONTEXT((short)2, "log_context"),
-    HANDLE((short)3, "handle");
+    HANDLE((short)3, "handle"),
+    ERROR_CODE((short)4, "errorCode"),
+    SQLSTATE((short)5, "SQLState");
 
     private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
 
@@ -57,6 +63,10 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
           return LOG_CONTEXT;
         case 3: // HANDLE
           return HANDLE;
+        case 4: // ERROR_CODE
+          return ERROR_CODE;
+        case 5: // SQLSTATE
+          return SQLSTATE;
         default:
           return null;
       }
@@ -97,6 +107,8 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
   }
 
   // isset id assignments
+  private static final int __ERRORCODE_ISSET_ID = 0;
+  private BitSet __isset_bit_vector = new BitSet(1);
 
   public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
   static {
@@ -107,11 +119,19 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
         new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING        , "LogContextId")));
     tmpMap.put(_Fields.HANDLE, new org.apache.thrift.meta_data.FieldMetaData("handle", org.apache.thrift.TFieldRequirementType.DEFAULT, 
         new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, QueryHandle.class)));
+    tmpMap.put(_Fields.ERROR_CODE, new org.apache.thrift.meta_data.FieldMetaData("errorCode", org.apache.thrift.TFieldRequirementType.OPTIONAL, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I32)));
+    tmpMap.put(_Fields.SQLSTATE, new org.apache.thrift.meta_data.FieldMetaData("SQLState", org.apache.thrift.TFieldRequirementType.OPTIONAL, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
     metaDataMap = Collections.unmodifiableMap(tmpMap);
     org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(BeeswaxException.class, metaDataMap);
   }
 
   public BeeswaxException() {
+    this.errorCode = 0;
+
+    this.SQLState = "     ";
+
   }
 
   public BeeswaxException(
@@ -129,6 +149,8 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
    * Performs a deep copy on <i>other</i>.
    */
   public BeeswaxException(BeeswaxException other) {
+    __isset_bit_vector.clear();
+    __isset_bit_vector.or(other.__isset_bit_vector);
     if (other.isSetMessage()) {
       this.message = other.message;
     }
@@ -138,6 +160,10 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
     if (other.isSetHandle()) {
       this.handle = new QueryHandle(other.handle);
     }
+    this.errorCode = other.errorCode;
+    if (other.isSetSQLState()) {
+      this.SQLState = other.SQLState;
+    }
   }
 
   public BeeswaxException deepCopy() {
@@ -149,6 +175,10 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
     this.message = null;
     this.log_context = null;
     this.handle = null;
+    this.errorCode = 0;
+
+    this.SQLState = "     ";
+
   }
 
   public String getMessage() {
@@ -223,6 +253,53 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
     }
   }
 
+  public int getErrorCode() {
+    return this.errorCode;
+  }
+
+  public BeeswaxException setErrorCode(int errorCode) {
+    this.errorCode = errorCode;
+    setErrorCodeIsSet(true);
+    return this;
+  }
+
+  public void unsetErrorCode() {
+    __isset_bit_vector.clear(__ERRORCODE_ISSET_ID);
+  }
+
+  /** Returns true if field errorCode is set (has been assigned a value) and false otherwise */
+  public boolean isSetErrorCode() {
+    return __isset_bit_vector.get(__ERRORCODE_ISSET_ID);
+  }
+
+  public void setErrorCodeIsSet(boolean value) {
+    __isset_bit_vector.set(__ERRORCODE_ISSET_ID, value);
+  }
+
+  public String getSQLState() {
+    return this.SQLState;
+  }
+
+  public BeeswaxException setSQLState(String SQLState) {
+    this.SQLState = SQLState;
+    return this;
+  }
+
+  public void unsetSQLState() {
+    this.SQLState = null;
+  }
+
+  /** Returns true if field SQLState is set (has been assigned a value) and false otherwise */
+  public boolean isSetSQLState() {
+    return this.SQLState != null;
+  }
+
+  public void setSQLStateIsSet(boolean value) {
+    if (!value) {
+      this.SQLState = null;
+    }
+  }
+
   public void setFieldValue(_Fields field, Object value) {
     switch (field) {
     case MESSAGE:
@@ -249,6 +326,22 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
       }
       break;
 
+    case ERROR_CODE:
+      if (value == null) {
+        unsetErrorCode();
+      } else {
+        setErrorCode((Integer)value);
+      }
+      break;
+
+    case SQLSTATE:
+      if (value == null) {
+        unsetSQLState();
+      } else {
+        setSQLState((String)value);
+      }
+      break;
+
     }
   }
 
@@ -263,6 +356,12 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
     case HANDLE:
       return getHandle();
 
+    case ERROR_CODE:
+      return Integer.valueOf(getErrorCode());
+
+    case SQLSTATE:
+      return getSQLState();
+
     }
     throw new IllegalStateException();
   }
@@ -280,6 +379,10 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
       return isSetLog_context();
     case HANDLE:
       return isSetHandle();
+    case ERROR_CODE:
+      return isSetErrorCode();
+    case SQLSTATE:
+      return isSetSQLState();
     }
     throw new IllegalStateException();
   }
@@ -324,6 +427,24 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
         return false;
     }
 
+    boolean this_present_errorCode = true && this.isSetErrorCode();
+    boolean that_present_errorCode = true && that.isSetErrorCode();
+    if (this_present_errorCode || that_present_errorCode) {
+      if (!(this_present_errorCode && that_present_errorCode))
+        return false;
+      if (this.errorCode != that.errorCode)
+        return false;
+    }
+
+    boolean this_present_SQLState = true && this.isSetSQLState();
+    boolean that_present_SQLState = true && that.isSetSQLState();
+    if (this_present_SQLState || that_present_SQLState) {
+      if (!(this_present_SQLState && that_present_SQLState))
+        return false;
+      if (!this.SQLState.equals(that.SQLState))
+        return false;
+    }
+
     return true;
   }
 
@@ -346,6 +467,16 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
     if (present_handle)
       builder.append(handle);
 
+    boolean present_errorCode = true && (isSetErrorCode());
+    builder.append(present_errorCode);
+    if (present_errorCode)
+      builder.append(errorCode);
+
+    boolean present_SQLState = true && (isSetSQLState());
+    builder.append(present_SQLState);
+    if (present_SQLState)
+      builder.append(SQLState);
+
     return builder.toHashCode();
   }
 
@@ -387,6 +518,26 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
         return lastComparison;
       }
     }
+    lastComparison = Boolean.valueOf(isSetErrorCode()).compareTo(typedOther.isSetErrorCode());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetErrorCode()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.errorCode, typedOther.errorCode);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
+    lastComparison = Boolean.valueOf(isSetSQLState()).compareTo(typedOther.isSetSQLState());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetSQLState()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.SQLState, typedOther.SQLState);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
     return 0;
   }
 
@@ -426,6 +577,21 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
             org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
           }
           break;
+        case 4: // ERROR_CODE
+          if (field.type == org.apache.thrift.protocol.TType.I32) {
+            this.errorCode = iprot.readI32();
+            setErrorCodeIsSet(true);
+          } else { 
+            org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
+          }
+          break;
+        case 5: // SQLSTATE
+          if (field.type == org.apache.thrift.protocol.TType.STRING) {
+            this.SQLState = iprot.readString();
+          } else { 
+            org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
+          }
+          break;
         default:
           org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
       }
@@ -456,6 +622,18 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
       this.handle.write(oprot);
       oprot.writeFieldEnd();
     }
+    if (isSetErrorCode()) {
+      oprot.writeFieldBegin(ERROR_CODE_FIELD_DESC);
+      oprot.writeI32(this.errorCode);
+      oprot.writeFieldEnd();
+    }
+    if (this.SQLState != null) {
+      if (isSetSQLState()) {
+        oprot.writeFieldBegin(SQLSTATE_FIELD_DESC);
+        oprot.writeString(this.SQLState);
+        oprot.writeFieldEnd();
+      }
+    }
     oprot.writeFieldStop();
     oprot.writeStructEnd();
   }
@@ -488,6 +666,22 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
       sb.append(this.handle);
     }
     first = false;
+    if (isSetErrorCode()) {
+      if (!first) sb.append(", ");
+      sb.append("errorCode:");
+      sb.append(this.errorCode);
+      first = false;
+    }
+    if (isSetSQLState()) {
+      if (!first) sb.append(", ");
+      sb.append("SQLState:");
+      if (this.SQLState == null) {
+        sb.append("null");
+      } else {
+        sb.append(this.SQLState);
+      }
+      first = false;
+    }
     sb.append(")");
     return sb.toString();
   }
@@ -506,6 +700,8 @@ public class BeeswaxException extends Exception implements org.apache.thrift.TBa
 
   private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
     try {
+      // it doesn't seem like you should have to do this, but java serialization is wacky, and doesn't call the default constructor.
+      __isset_bit_vector = new BitSet(1);
       read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
     } catch (org.apache.thrift.TException te) {
       throw new java.io.IOException(te);

Tiedoston diff-näkymää rajattu, sillä se on liian suuri
+ 435 - 132
apps/beeswax/java/src/main/gen-java/com/cloudera/beeswax/api/BeeswaxService.java


+ 172 - 24
apps/beeswax/java/src/main/java/com/cloudera/beeswax/BeeswaxServiceImpl.java

@@ -60,6 +60,7 @@ import org.apache.hadoop.hive.ql.plan.FetchWork;
 import org.apache.hadoop.hive.ql.plan.TableDesc;
 import org.apache.hadoop.hive.ql.processors.CommandProcessor;
 import org.apache.hadoop.hive.ql.processors.CommandProcessorFactory;
+import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
 import org.apache.hadoop.hive.ql.QueryPlan;
 import org.apache.hadoop.hive.ql.session.SessionState;
 import org.apache.hadoop.hive.serde.Constants;
@@ -331,11 +332,47 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
       }
     }
 
+    /**
+     * Run query. Updates state.
+     * @throws BeeswaxException
+     */
+    public void run() throws BeeswaxException {
+      CommandProcessorResponse response;
+
+      synchronized (this) {
+        assertState(QueryState.INITIALIZED);
+        state = QueryState.RUNNING;
+      }
+      try {
+        response = driver.run(query.query);
+      } catch (Exception e) {
+        throw new BeeswaxException(getErrorStreamAsString(), logContext.getName(), this.handle);
+      }
+      if (response.getResponseCode() != 0) {
+         throwExceptionWithSqlErrors(new BeeswaxException(response.getErrorMessage(),
+             logContext.getName(), this.handle),
+             response.getResponseCode(), response.getSQLState());
+      } else {
+        synchronized (this) {
+          state = QueryState.FINISHED;
+        }
+      }
+    }
+
+    // Add the SQLCODE and SQLState to the BeeswaxException. Thrift doesn't provide
+    // exception constructors with optional members hence it's done here.
+    private void throwExceptionWithSqlErrors(BeeswaxException ex, int errCode,
+            String sqlState) throws BeeswaxException {
+      ex.setSQLState(sqlState);
+      ex.setErrorCode(errCode);
+      throw ex;
+    }
+
     public void bringUp() {
       SessionState.start(this.sessionState);
     }
 
-    private void materializeResults(Results r, boolean startOver)
+    private void materializeResults(Results r, boolean startOver, int fetchSize)
     throws IOException, CommandNeedRetryException {
       if (driver.getPlan().getFetchTask() == null) {
         // This query is never going to return anything.
@@ -359,6 +396,9 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
 
       ArrayList<String> v = new ArrayList<String>();
       r.setData(v);
+      if (fetchSize > 0) {
+        driver.setMaxRows(fetchSize);
+      }
       r.has_more = driver.getResults(v);
       r.start_row = startRow;
       startRow += v.size();
@@ -454,7 +494,8 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
       return new QueryExplanation(sb.toString());
     }
 
-    public Results fetch(boolean fromBeginning) throws BeeswaxException {
+    public Results fetch(boolean fromBeginning, int fetchSize)
+                throws BeeswaxException {
       this.atime = System.currentTimeMillis();
       Results r = new Results();
       // Only one person can access a running query at a time.
@@ -467,7 +508,7 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
           bringUp();
           r.ready = true;
           try {
-            materializeResults(r, fromBeginning);
+            materializeResults(r, fromBeginning, fetchSize);
           } catch (Exception e) {
             throw new BeeswaxException(e.toString(), logContext.getName(), handle);
           } finally {
@@ -485,6 +526,19 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
       return r;
     }
 
+    /**
+     * Close the query. Free the resource held by driver
+     */
+    public int close() {
+      int ret = -1;
+      try {
+        ret = driver.close();
+      } catch (Exception e) {
+        LOG.error("Exception while closing query", e);
+      }
+      return ret;
+    }
+
     /** Store the first exception we see. */
     private void saveException(Throwable t) {
       synchronized (this) {
@@ -547,6 +601,9 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
    * (We don't care whether the notification handling fails or succeeds.)
    */
   void notifyDone(RunningQueryState state) {
+    if (notifyUrl == null) {
+      return;
+    }
     QueryHandle handle = state.getQueryHandle();
     if (handle == null) {
       LOG.error("Finished execution of a query without a handle: " + state.toString());
@@ -596,27 +653,31 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
     this.runningQueries = new ConcurrentHashMap<String, RunningQueryState>();
     this.queryLifetime = queryLifetime;
 
-    String protocol;
-    if (dtHttps) {
-      protocol = "https";
-      try {
-        // Disable SSL verification. HUE cert may be signed by untrusted CA.
-        SSLContext sslcontext = SSLContext.getInstance("SSL");
-        sslcontext.init(null,
-                        new DummyX509TrustManager[] { new DummyX509TrustManager() },
-                        new SecureRandom());
-        HttpsURLConnection.setDefaultSSLSocketFactory(sslcontext.getSocketFactory());
-      } catch (NoSuchAlgorithmException ex) {
-        LOG.warn("Failed to disable SSL certificate check " + ex);
-      } catch (KeyManagementException ex) {
-        LOG.warn("Failed to disable SSL certificate check " + ex);
-      }
-      DummyHostnameVerifier dummy = new DummyHostnameVerifier();
-      HttpsURLConnection.setDefaultHostnameVerifier(dummy);
+    if (dtPort == -1) {
+      this.notifyUrl = null;
     } else {
-      protocol = "http";
+      String protocol;
+      if (dtHttps) {
+        try {
+          // Disable SSL verification. HUE cert may be signed by untrusted CA.
+          SSLContext sslcontext = SSLContext.getInstance("SSL");
+          sslcontext.init(null,
+                          new DummyX509TrustManager[] { new DummyX509TrustManager() },
+                          new SecureRandom());
+          HttpsURLConnection.setDefaultSSLSocketFactory(sslcontext.getSocketFactory());
+        } catch (NoSuchAlgorithmException ex) {
+          LOG.warn("Failed to disable SSL certificate check " + ex);
+        } catch (KeyManagementException ex) {
+          LOG.warn("Failed to disable SSL certificate check " + ex);
+        }
+        DummyHostnameVerifier dummy = new DummyHostnameVerifier();
+        HttpsURLConnection.setDefaultHostnameVerifier(dummy);
+        protocol = "https";
+      } else {
+        protocol = "http";
+      }
+      this.notifyUrl = protocol + "://" + dtHost + ":" + dtPort + NOTIFY_URL_BASE;
     }
-    this.notifyUrl = protocol + "://" + dtHost + ":" + dtPort + NOTIFY_URL_BASE;
 
     // A daemon thread that periodically evict stale RunningQueryState objects
     Thread evicter = new Thread(new Runnable() {
@@ -724,6 +785,53 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
     }
   }
 
+  /**
+   * Run a query synchronously and return a handle (QueryHandle).
+   */
+  @Override
+  public QueryHandle executeAndWait(final Query query, String clientUuid)
+      throws BeeswaxException {
+    // First, create an id and reset the LogContext
+    String queryUuid = UUID.randomUUID().toString();
+    String contextUuid;
+
+    LOG.info("got context " + clientUuid);
+    if (clientUuid.isEmpty()) {
+      contextUuid = queryUuid;
+    } else {
+      contextUuid = clientUuid;
+    }
+    LOG.info("running query " + query.query + " context " + contextUuid);
+    final QueryHandle handle = new QueryHandle(queryUuid, contextUuid);
+    final LogContext lc = LogContext.registerCurrentThread(handle.log_context);
+    lc.resetLog();
+
+    // Make an administrative record
+    final RunningQueryState state = new RunningQueryState(query, lc);
+    try {
+      return doWithState(state,
+          new PrivilegedExceptionAction<QueryHandle>() {
+            public QueryHandle run() throws Exception {
+              state.setQueryHandle(handle);
+              runningQueries.put(handle.id, state);
+              state.initialize();
+              try {
+                state.run();
+              } catch (BeeswaxException perr) {
+                state.saveException(perr);
+                throw perr;
+              } catch (Throwable t) {
+                state.saveException(t);
+                throw new BeeswaxException(t.toString(), handle.log_context, handle);
+              }
+              return handle;
+            }
+          });
+    } catch (BeeswaxException e) {
+      throw e;
+    }
+  }
+
   /**
    * Verify that the handle data is not null.
    */
@@ -781,9 +889,11 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
    *
    * @param handle  The handle from query()
    * @param fromBeginning  If true, rewind to the first row. Otherwise fetch from last position.
+   * @param fetchSize  Number of rows to return with this fetch
    */
   @Override
-  public Results fetch(final QueryHandle handle, final boolean fromBeginning)
+  public Results fetch(final QueryHandle handle, final boolean fromBeginning,
+            final int fetchSize)
       throws QueryNotFoundException, BeeswaxException {
     LogContext.unregisterCurrentThread();
     validateHandle(handle);
@@ -796,7 +906,7 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
               if (state == null) {
                 throw new QueryNotFoundException();
               }
-              return state.fetch(fromBeginning);
+              return state.fetch(fromBeginning, fetchSize);
             }
           });
     } catch (BeeswaxException e) {
@@ -913,4 +1023,42 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
   public void setQueryLifetime(long queryLifetime) {
     this.queryLifetime = queryLifetime;
   }
+
+  /*
+   * close the query
+   */
+  @Override
+  public void close(QueryHandle handle)
+    throws QueryNotFoundException, BeeswaxException {
+    LogContext.unregisterCurrentThread();
+    validateHandle(handle);
+    LogContext.registerCurrentThread(handle.log_context);
+    final RunningQueryState state = runningQueries.get(handle.id);
+    try {
+      doWithState(state,
+          new PrivilegedExceptionAction<Integer>() {
+            public Integer run() throws Exception {
+              if (state == null) {
+                throw new QueryNotFoundException();
+              }
+              return state.close();
+            }
+          });
+    } catch (BeeswaxException e) {
+      throw e;
+    }
+    runningQueries.remove(handle.id);
+  }
+
+  @Override
+  public void clean(String contextName) {
+    LogContext.unregisterCurrentThread();
+    try {
+      if (LogContext.destroyContext(contextName) == false) {
+        throw new QueryNotFoundException();
+      }
+    } catch (Exception e) {
+      LOG.error("Error freeing query resources", e);
+    }
+  }
 }

+ 72 - 6
apps/beeswax/java/src/main/java/com/cloudera/beeswax/Server.java

@@ -16,9 +16,12 @@
 package com.cloudera.beeswax;
 
 import java.io.IOException;
+import java.net.Socket;
 
 import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.MissingOptionException;
 import org.apache.commons.cli.Option;
+import org.apache.commons.cli.OptionGroup;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.ParseException;
 import org.apache.commons.cli.PosixParser;
@@ -31,11 +34,16 @@ import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore;
 import org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore.Iface;
 import org.apache.hadoop.hive.ql.Driver;
+import org.apache.hadoop.security.SaslRpcServer;
+import org.apache.hadoop.security.SecurityUtil;
+import org.apache.hadoop.security.SaslRpcServer.AuthMethod;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.log4j.Logger;
 import org.apache.thrift.TProcessor;
 import org.apache.thrift.protocol.TBinaryProtocol;
 import org.apache.thrift.server.TServer;
 import org.apache.thrift.server.TThreadPoolServer;
+import org.apache.thrift.transport.TSaslServerTransport;
 import org.apache.thrift.transport.TServerSocket;
 import org.apache.thrift.transport.TServerTransport;
 import org.apache.thrift.transport.TTransportException;
@@ -59,6 +67,9 @@ public class Server {
   private static int dtPort = -1;
   private static boolean dtHttps = false;
   private static long qlifetime = BeeswaxServiceImpl.RUNNING_QUERY_LIFETIME;
+  private static boolean useSasl = false;
+  private static String principalConf;
+  private static String keytabFile;
 
   /**
    * Parse command line options.
@@ -68,6 +79,8 @@ public class Server {
    */
   private static void parseArgs(String[] args) throws ParseException {
     Options options = new Options();
+    OptionGroup dtOptions = new OptionGroup();
+    dtOptions.setRequired(true);
 
     Option metastoreOpt = new Option("m", "metastore", true, "port to use for metastore");
     metastoreOpt.setRequired(false);
@@ -83,23 +96,34 @@ public class Server {
     options.addOption(beeswaxOpt);
 
     Option dtHostOpt = new Option("h", "desktop-host", true, "host running desktop");
-    dtHostOpt.setRequired(true);
+    dtHostOpt.setRequired(false);
     options.addOption(dtHostOpt);
 
     Option dtHttpsOpt = new Option("s", "desktop-https", true, "desktop is running https");
     options.addOption(dtHttpsOpt);
 
     Option dtPortOpt = new Option("p", "desktop-port", true, "port used by desktop");
-    dtPortOpt.setRequired(true);
-    options.addOption(dtPortOpt);
+    dtPortOpt.setRequired(false);
+    dtOptions.addOption(dtPortOpt);
 
     Option superUserOpt = new Option("u", "superuser", true,
                                     "Username of Hadoop superuser (default: hadoop)");
     superUserOpt.setRequired(false);
     options.addOption(superUserOpt);
 
+    Option noDesktopOpt = new Option("n", "no-desktop", false, "no desktop used");
+    dtOptions.addOption(noDesktopOpt);
+
+    Option kPrincipalOpt = new Option("c", "principalConf", true, "Pricipal configuration");
+    options.addOption(kPrincipalOpt);
+
+    Option kTabOpt = new Option("k", "keytab", true, "keytab file");
+    options.addOption(kTabOpt);
+
+    options.addOptionGroup(dtOptions); // make "n" and "p" mutually exclusive
     PosixParser parser = new PosixParser();
     CommandLine cmd = parser.parse(options, args);
+
     if (!cmd.getArgList().isEmpty()) {
       throw new ParseException("Unexpected extra arguments: " + cmd.getArgList());
     }
@@ -119,6 +143,11 @@ public class Server {
         dtHttps = true;
       } else if (opt.getOpt() == "l") {
         qlifetime = Long.valueOf(opt.getValue());
+      } else if (opt.getOpt() == "k") {
+        keytabFile = opt.getValue();
+        useSasl = true;
+      } else if (opt.getOpt() == "c") {
+        principalConf = opt.getValue();
       }
     }
   }
@@ -196,14 +225,51 @@ public class Server {
     BeeswaxService.Iface impl = new BeeswaxServiceImpl(dtHost, dtPort, dtHttps,
         qlifetime);
     Processor processor = new BeeswaxService.Processor(impl);
+    TTransportFactory transFactory;
+
+    if (useSasl) {
+      if (keytabFile == null || keytabFile.isEmpty()) {
+        throw new IllegalArgumentException("No keytab specified");
+      }
+      if (principalConf == null || principalConf.isEmpty()) {
+        throw new IllegalArgumentException("No principal specified");
+      }
+
+      // Login from the keytab
+      String kerberosName;
+      try {
+        kerberosName =
+          SecurityUtil.getServerPrincipal(principalConf, "0.0.0.0");
+        UserGroupInformation.loginUserFromKeytab(
+            kerberosName, keytabFile);
+        kerberosName = UserGroupInformation.getCurrentUser().getUserName();
+      } catch (IOException e) {
+        throw new IllegalArgumentException("couldn't setup authentication subsystem", e);
+      }
+      final String names[] = SaslRpcServer.splitKerberosName(kerberosName);
+      if (names.length != 3) {
+        throw new IllegalArgumentException("Kerberos principal should have 3 parts: " + kerberosName);
+      }
+      transFactory = new TSaslServerTransport.Factory(AuthMethod.KERBEROS.getMechanismName(),
+          names[0], names[1],  // two parts of kerberos principal
+          SaslRpcServer.SASL_PROPS,
+          new SaslRpcServer.SaslGssCallbackHandler());
+    } else {
+      transFactory = new TTransportFactory();
+    }
+
     TThreadPoolServer.Args args = new TThreadPoolServer.Args(serverTransport)
         .processor(processor)
         .protocolFactory(new TBinaryProtocol.Factory())
-        .transportFactory(new TTransportFactory());
+        .transportFactory(transFactory);
     TServer server = new TThreadPoolServer(args);
-    LOG.info("Starting beeswax server on port " + port + ", talking back to Desktop at " +
-             dtHost + ":" + dtPort + ", lifetime of queries set to " + qlifetime);
 
+    if (dtPort != -1) {
+      LOG.info("Starting beeswax server on port " + port + ", talking back to Desktop at "
+                + dtHost + ":" + dtPort);
+    } else {
+      LOG.info("Starting beeswax server on port " + port);
+    }
     server.serve();
   }
 

+ 2 - 0
apps/beeswax/regenerate_thrift.sh

@@ -44,3 +44,5 @@ EOF
 
 # This is based on thirdparty.
 # thrift -r --gen py:new_style -o ../ ../../../../ext/thirdparty/py/thrift/contrib/fb303/if/fb303.thrift
+# C++ compilation for ODBC
+#thrift -I thrift/include  --gen cpp -o ./ thrift/beeswax.thrift

+ 1 - 1
apps/beeswax/src/beeswax/data_export.py

@@ -77,7 +77,7 @@ def data_generator(query_model, formatter):
   while True:
     # Make sure that we have the next batch of ready results
     while results is None or not results.ready:
-      results = db_utils.db_client().fetch(handle, start_over=is_first_row)
+      results = db_utils.db_client().fetch(handle, start_over=is_first_row, fetch_size=-1)
       if not results.ready:
         time.sleep(_DATA_WAIT_SLEEP)
 

+ 1 - 1
apps/beeswax/src/beeswax/db_utils.py

@@ -99,7 +99,7 @@ def wait_for_results(query_history, timeout_sec=30.0):
   curr = time.time()
   end = curr + timeout_sec
   while curr <= end:
-    results = db_client().fetch(handle, START_OVER)
+    results = db_client().fetch(handle, START_OVER, fetch_size=-1)
     if results.ready:
       return results
     time.sleep(SLEEP_INTERVAL)

+ 46 - 0
apps/beeswax/src/beeswax/tests.py

@@ -46,6 +46,7 @@ from beeswax.views import parse_results, collapse_whitespace
 from beeswax.test_base import make_query, wait_for_query_to_finish, verify_history
 from beeswax.test_base import BeeswaxSampleProvider
 from beeswaxd import BeeswaxService
+from beeswaxd import ttypes
 
 LOG = logging.getLogger(__name__)
 CSV_LINK_PAT = re.compile('/beeswax/download/\d+/csv')
@@ -203,6 +204,7 @@ for x in sys.stdin:
     assert_equal(0, len(response.context["hadoop_jobs"]), "Shouldn't have found jobs.")
     self._verify_query_state(beeswax.models.QueryHistory.STATE.available)
 
+
     # Query multi-page request
     QUERY = """
       SELECT * FROM test
@@ -276,6 +278,48 @@ for x in sys.stdin:
     resp = self.client.get('/beeswax/watch/%s' % (id,), follow=True)
     check_error_in_response(resp)
 
+  def test_sync_query_exec(self):
+    # Execute Query Synchronously, set fetch size and fetch results
+    # verify the size of resultset,
+    QUERY = """
+      SELECT foo FROM test;
+    """
+
+    query_msg = BeeswaxService.Query()
+    query_msg.query = """
+      SELECT foo FROM test
+    """
+    query_msg.configuration = []
+    query_msg.hadoop_user = "test"
+    handle = beeswax.db_utils.db_client().executeAndWait(query_msg, "")
+ 
+    results = beeswax.db_utils.db_client().fetch(handle, True, 5)
+    row_list = list(parse_results(results.data))
+    assert_equal(len(row_list), 5)
+
+    beeswax.db_utils.db_client().close(handle)
+    beeswax.db_utils.db_client().clean(handle.log_context)
+
+
+  def test_sync_query_error(self):
+    # Execute incorrect Query , verify the error code and sqlstate
+    QUERY = """
+      SELECT foo FROM test;
+    """
+
+    query_msg = BeeswaxService.Query()
+    query_msg.query = """
+      SELECT FROM zzzzz
+    """
+    query_msg.configuration = []
+    query_msg.hadoop_user = "test"
+    try:
+      handle = beeswax.db_utils.db_client().executeAndWait(query_msg, "")
+    except ttypes.BeeswaxException, bex:
+      assert_equal(bex.errorCode, 11)
+      assert_equal(bex.SQLState, "42000")
+
+
   def test_parameterization(self):
     """
     Test parameterization
@@ -1075,3 +1119,5 @@ def test_collapse_whitespace():
   assert_equal("", collapse_whitespace("\t\n\n  \n\t \n"))
   assert_equal("x", collapse_whitespace("\t\nx\n  \n\t \n"))
   assert_equal("x y", collapse_whitespace("\t\nx\n  \ny\t \n"))
+
+

+ 1 - 1
apps/beeswax/src/beeswax/views.py

@@ -896,7 +896,7 @@ def view_results(request, id, first_row=0):
 
   # Retrieve query results
   try:
-    results = db_utils.db_client().fetch(handle, start_over)
+    results = db_utils.db_client().fetch(handle, start_over, -1)
     assert results.ready, 'Trying to display result that is not yet ready. Query id %s' % (id,)
     # We display the "Download" button only when we know
     # that there are results:

+ 24 - 3
apps/beeswax/thrift/beeswax.thrift

@@ -86,7 +86,9 @@ exception BeeswaxException {
   // Use get_log(log_context) to retrieve any log related to this exception
   2: LogContextId log_context,
   // (Optional) The QueryHandle that caused this exception
-  3: QueryHandle handle
+  3: QueryHandle handle,
+  4: optional i32 errorCode = 0,
+  5: optional string SQLState = "     "
 }
 
 exception QueryNotFoundException {
@@ -105,6 +107,12 @@ service BeeswaxService {
    */
   QueryHandle query(1:Query query) throws(1:BeeswaxException error),
 
+  /**
+   * run a query synchronously and return a handle (QueryHandle).
+   */
+  QueryHandle executeAndWait(1:Query query, 2:LogContextId clientCtx) 
+                        throws(1:BeeswaxException error),
+
   /**
    * Get the query plan for a query.
    */
@@ -113,9 +121,11 @@ service BeeswaxService {
 
   /**
    * Get the results of a query. This is non-blocking. Caller should check
-   * Results.ready to determine if the results are in yet.
+   * Results.ready to determine if the results are in yet. The call requests
+   * the batch size of fetch.
    */
-  Results fetch(1:QueryHandle query_id, 2:bool start_over) throws(1:QueryNotFoundException error, 2:BeeswaxException error2),
+  Results fetch(1:QueryHandle query_id, 2:bool start_over, 3:i32 fetch_size=-1) 
+              throws(1:QueryNotFoundException error, 2:BeeswaxException error2),
 
   /**
    * Get the state of the query
@@ -148,4 +158,15 @@ service BeeswaxService {
    * Returns "default" configuration.
    */
   list<ConfigVariable> get_default_configuration(1:bool include_hadoop)
+
+  /*
+   * closes the query with given handle
+   */
+  void close(1:QueryHandle handle) throws(1:QueryNotFoundException error, 
+                            2:BeeswaxException error2)
+
+  /*
+   * clean the log context for given id 
+   */
+  void clean(1:LogContextId log_context)
 }

Kaikkia tiedostoja ei voida näyttää, sillä liian monta tiedostoa muuttui tässä diffissä