You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2015/01/23 20:59:24 UTC
svn commit: r1654355 [9/27] - in /hive/branches/llap: ./
beeline/src/java/org/apache/hive/beeline/
cli/src/java/org/apache/hadoop/hive/cli/
common/src/java/org/apache/hadoop/hive/common/
common/src/java/org/apache/hadoop/hive/conf/ data/conf/ data/conf...
Modified: hive/branches/llap/metastore/src/gen/thrift/gen-py/hive_metastore/ttypes.py
URL: http://svn.apache.org/viewvc/hive/branches/llap/metastore/src/gen/thrift/gen-py/hive_metastore/ttypes.py?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/metastore/src/gen/thrift/gen-py/hive_metastore/ttypes.py (original)
+++ hive/branches/llap/metastore/src/gen/thrift/gen-py/hive_metastore/ttypes.py Fri Jan 23 19:59:11 2015
@@ -8143,6 +8143,341 @@ class ShowCompactResponse:
def __ne__(self, other):
return not (self == other)
+class NotificationEventRequest:
+ """
+ Attributes:
+ - lastEvent
+ - maxEvents
+ """
+
+ thrift_spec = (
+ None, # 0
+ (1, TType.I64, 'lastEvent', None, None, ), # 1
+ (2, TType.I32, 'maxEvents', None, None, ), # 2
+ )
+
+ def __init__(self, lastEvent=None, maxEvents=None,):
+ self.lastEvent = lastEvent
+ self.maxEvents = maxEvents
+
+ 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.I64:
+ self.lastEvent = iprot.readI64();
+ else:
+ iprot.skip(ftype)
+ elif fid == 2:
+ if ftype == TType.I32:
+ self.maxEvents = iprot.readI32();
+ 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('NotificationEventRequest')
+ if self.lastEvent is not None:
+ oprot.writeFieldBegin('lastEvent', TType.I64, 1)
+ oprot.writeI64(self.lastEvent)
+ oprot.writeFieldEnd()
+ if self.maxEvents is not None:
+ oprot.writeFieldBegin('maxEvents', TType.I32, 2)
+ oprot.writeI32(self.maxEvents)
+ oprot.writeFieldEnd()
+ oprot.writeFieldStop()
+ oprot.writeStructEnd()
+
+ def validate(self):
+ if self.lastEvent is None:
+ raise TProtocol.TProtocolException(message='Required field lastEvent is unset!')
+ 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 NotificationEvent:
+ """
+ Attributes:
+ - eventId
+ - eventTime
+ - eventType
+ - dbName
+ - tableName
+ - message
+ """
+
+ thrift_spec = (
+ None, # 0
+ (1, TType.I64, 'eventId', None, None, ), # 1
+ (2, TType.I32, 'eventTime', None, None, ), # 2
+ (3, TType.STRING, 'eventType', None, None, ), # 3
+ (4, TType.STRING, 'dbName', None, None, ), # 4
+ (5, TType.STRING, 'tableName', None, None, ), # 5
+ (6, TType.STRING, 'message', None, None, ), # 6
+ )
+
+ def __init__(self, eventId=None, eventTime=None, eventType=None, dbName=None, tableName=None, message=None,):
+ self.eventId = eventId
+ self.eventTime = eventTime
+ self.eventType = eventType
+ self.dbName = dbName
+ self.tableName = tableName
+ self.message = message
+
+ 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.I64:
+ self.eventId = iprot.readI64();
+ else:
+ iprot.skip(ftype)
+ elif fid == 2:
+ if ftype == TType.I32:
+ self.eventTime = iprot.readI32();
+ else:
+ iprot.skip(ftype)
+ elif fid == 3:
+ if ftype == TType.STRING:
+ self.eventType = iprot.readString();
+ else:
+ iprot.skip(ftype)
+ elif fid == 4:
+ if ftype == TType.STRING:
+ self.dbName = iprot.readString();
+ else:
+ iprot.skip(ftype)
+ elif fid == 5:
+ if ftype == TType.STRING:
+ self.tableName = iprot.readString();
+ else:
+ iprot.skip(ftype)
+ elif fid == 6:
+ if ftype == TType.STRING:
+ self.message = 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('NotificationEvent')
+ if self.eventId is not None:
+ oprot.writeFieldBegin('eventId', TType.I64, 1)
+ oprot.writeI64(self.eventId)
+ oprot.writeFieldEnd()
+ if self.eventTime is not None:
+ oprot.writeFieldBegin('eventTime', TType.I32, 2)
+ oprot.writeI32(self.eventTime)
+ oprot.writeFieldEnd()
+ if self.eventType is not None:
+ oprot.writeFieldBegin('eventType', TType.STRING, 3)
+ oprot.writeString(self.eventType)
+ oprot.writeFieldEnd()
+ if self.dbName is not None:
+ oprot.writeFieldBegin('dbName', TType.STRING, 4)
+ oprot.writeString(self.dbName)
+ oprot.writeFieldEnd()
+ if self.tableName is not None:
+ oprot.writeFieldBegin('tableName', TType.STRING, 5)
+ oprot.writeString(self.tableName)
+ oprot.writeFieldEnd()
+ if self.message is not None:
+ oprot.writeFieldBegin('message', TType.STRING, 6)
+ oprot.writeString(self.message)
+ oprot.writeFieldEnd()
+ oprot.writeFieldStop()
+ oprot.writeStructEnd()
+
+ def validate(self):
+ if self.eventId is None:
+ raise TProtocol.TProtocolException(message='Required field eventId is unset!')
+ if self.eventTime is None:
+ raise TProtocol.TProtocolException(message='Required field eventTime is unset!')
+ if self.eventType is None:
+ raise TProtocol.TProtocolException(message='Required field eventType is unset!')
+ if self.message is None:
+ raise TProtocol.TProtocolException(message='Required field message is unset!')
+ 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 NotificationEventResponse:
+ """
+ Attributes:
+ - events
+ """
+
+ thrift_spec = (
+ None, # 0
+ (1, TType.LIST, 'events', (TType.STRUCT,(NotificationEvent, NotificationEvent.thrift_spec)), None, ), # 1
+ )
+
+ def __init__(self, events=None,):
+ self.events = events
+
+ 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.LIST:
+ self.events = []
+ (_etype437, _size434) = iprot.readListBegin()
+ for _i438 in xrange(_size434):
+ _elem439 = NotificationEvent()
+ _elem439.read(iprot)
+ self.events.append(_elem439)
+ iprot.readListEnd()
+ 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('NotificationEventResponse')
+ if self.events is not None:
+ oprot.writeFieldBegin('events', TType.LIST, 1)
+ oprot.writeListBegin(TType.STRUCT, len(self.events))
+ for iter440 in self.events:
+ iter440.write(oprot)
+ oprot.writeListEnd()
+ oprot.writeFieldEnd()
+ oprot.writeFieldStop()
+ oprot.writeStructEnd()
+
+ def validate(self):
+ if self.events is None:
+ raise TProtocol.TProtocolException(message='Required field events is unset!')
+ 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 CurrentNotificationEventId:
+ """
+ Attributes:
+ - eventId
+ """
+
+ thrift_spec = (
+ None, # 0
+ (1, TType.I64, 'eventId', None, None, ), # 1
+ )
+
+ def __init__(self, eventId=None,):
+ self.eventId = eventId
+
+ 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.I64:
+ self.eventId = iprot.readI64();
+ 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('CurrentNotificationEventId')
+ if self.eventId is not None:
+ oprot.writeFieldBegin('eventId', TType.I64, 1)
+ oprot.writeI64(self.eventId)
+ oprot.writeFieldEnd()
+ oprot.writeFieldStop()
+ oprot.writeStructEnd()
+
+ def validate(self):
+ if self.eventId is None:
+ raise TProtocol.TProtocolException(message='Required field eventId is unset!')
+ 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 MetaException(TException):
"""
Attributes:
Modified: hive/branches/llap/metastore/src/gen/thrift/gen-rb/hive_metastore_types.rb
URL: http://svn.apache.org/viewvc/hive/branches/llap/metastore/src/gen/thrift/gen-rb/hive_metastore_types.rb?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/metastore/src/gen/thrift/gen-rb/hive_metastore_types.rb (original)
+++ hive/branches/llap/metastore/src/gen/thrift/gen-rb/hive_metastore_types.rb Fri Jan 23 19:59:11 2015
@@ -1986,6 +1986,89 @@ class ShowCompactResponse
::Thrift::Struct.generate_accessors self
end
+class NotificationEventRequest
+ include ::Thrift::Struct, ::Thrift::Struct_Union
+ LASTEVENT = 1
+ MAXEVENTS = 2
+
+ FIELDS = {
+ LASTEVENT => {:type => ::Thrift::Types::I64, :name => 'lastEvent'},
+ MAXEVENTS => {:type => ::Thrift::Types::I32, :name => 'maxEvents', :optional => true}
+ }
+
+ def struct_fields; FIELDS; end
+
+ def validate
+ raise ::Thrift::ProtocolException.new(::Thrift::ProtocolException::UNKNOWN, 'Required field lastEvent is unset!') unless @lastEvent
+ end
+
+ ::Thrift::Struct.generate_accessors self
+end
+
+class NotificationEvent
+ include ::Thrift::Struct, ::Thrift::Struct_Union
+ EVENTID = 1
+ EVENTTIME = 2
+ EVENTTYPE = 3
+ DBNAME = 4
+ TABLENAME = 5
+ MESSAGE = 6
+
+ FIELDS = {
+ EVENTID => {:type => ::Thrift::Types::I64, :name => 'eventId'},
+ EVENTTIME => {:type => ::Thrift::Types::I32, :name => 'eventTime'},
+ EVENTTYPE => {:type => ::Thrift::Types::STRING, :name => 'eventType'},
+ DBNAME => {:type => ::Thrift::Types::STRING, :name => 'dbName', :optional => true},
+ TABLENAME => {:type => ::Thrift::Types::STRING, :name => 'tableName', :optional => true},
+ MESSAGE => {:type => ::Thrift::Types::STRING, :name => 'message'}
+ }
+
+ def struct_fields; FIELDS; end
+
+ def validate
+ raise ::Thrift::ProtocolException.new(::Thrift::ProtocolException::UNKNOWN, 'Required field eventId is unset!') unless @eventId
+ raise ::Thrift::ProtocolException.new(::Thrift::ProtocolException::UNKNOWN, 'Required field eventTime is unset!') unless @eventTime
+ raise ::Thrift::ProtocolException.new(::Thrift::ProtocolException::UNKNOWN, 'Required field eventType is unset!') unless @eventType
+ raise ::Thrift::ProtocolException.new(::Thrift::ProtocolException::UNKNOWN, 'Required field message is unset!') unless @message
+ end
+
+ ::Thrift::Struct.generate_accessors self
+end
+
+class NotificationEventResponse
+ include ::Thrift::Struct, ::Thrift::Struct_Union
+ EVENTS = 1
+
+ FIELDS = {
+ EVENTS => {:type => ::Thrift::Types::LIST, :name => 'events', :element => {:type => ::Thrift::Types::STRUCT, :class => ::NotificationEvent}}
+ }
+
+ def struct_fields; FIELDS; end
+
+ def validate
+ raise ::Thrift::ProtocolException.new(::Thrift::ProtocolException::UNKNOWN, 'Required field events is unset!') unless @events
+ end
+
+ ::Thrift::Struct.generate_accessors self
+end
+
+class CurrentNotificationEventId
+ include ::Thrift::Struct, ::Thrift::Struct_Union
+ EVENTID = 1
+
+ FIELDS = {
+ EVENTID => {:type => ::Thrift::Types::I64, :name => 'eventId'}
+ }
+
+ def struct_fields; FIELDS; end
+
+ def validate
+ raise ::Thrift::ProtocolException.new(::Thrift::ProtocolException::UNKNOWN, 'Required field eventId is unset!') unless @eventId
+ end
+
+ ::Thrift::Struct.generate_accessors self
+end
+
class MetaException < ::Thrift::Exception
include ::Thrift::Struct, ::Thrift::Struct_Union
def initialize(message=nil)
Modified: hive/branches/llap/metastore/src/gen/thrift/gen-rb/thrift_hive_metastore.rb
URL: http://svn.apache.org/viewvc/hive/branches/llap/metastore/src/gen/thrift/gen-rb/thrift_hive_metastore.rb?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/metastore/src/gen/thrift/gen-rb/thrift_hive_metastore.rb (original)
+++ hive/branches/llap/metastore/src/gen/thrift/gen-rb/thrift_hive_metastore.rb Fri Jan 23 19:59:11 2015
@@ -1977,6 +1977,36 @@ module ThriftHiveMetastore
raise ::Thrift::ApplicationException.new(::Thrift::ApplicationException::MISSING_RESULT, 'show_compact failed: unknown result')
end
+ def get_next_notification(rqst)
+ send_get_next_notification(rqst)
+ return recv_get_next_notification()
+ end
+
+ def send_get_next_notification(rqst)
+ send_message('get_next_notification', Get_next_notification_args, :rqst => rqst)
+ end
+
+ def recv_get_next_notification()
+ result = receive_message(Get_next_notification_result)
+ return result.success unless result.success.nil?
+ raise ::Thrift::ApplicationException.new(::Thrift::ApplicationException::MISSING_RESULT, 'get_next_notification failed: unknown result')
+ end
+
+ def get_current_notificationEventId()
+ send_get_current_notificationEventId()
+ return recv_get_current_notificationEventId()
+ end
+
+ def send_get_current_notificationEventId()
+ send_message('get_current_notificationEventId', Get_current_notificationEventId_args)
+ end
+
+ def recv_get_current_notificationEventId()
+ result = receive_message(Get_current_notificationEventId_result)
+ return result.success unless result.success.nil?
+ raise ::Thrift::ApplicationException.new(::Thrift::ApplicationException::MISSING_RESULT, 'get_current_notificationEventId failed: unknown result')
+ end
+
end
class Processor < ::FacebookService::Processor
@@ -3493,6 +3523,20 @@ module ThriftHiveMetastore
write_result(result, oprot, 'show_compact', seqid)
end
+ def process_get_next_notification(seqid, iprot, oprot)
+ args = read_args(iprot, Get_next_notification_args)
+ result = Get_next_notification_result.new()
+ result.success = @handler.get_next_notification(args.rqst)
+ write_result(result, oprot, 'get_next_notification', seqid)
+ end
+
+ def process_get_current_notificationEventId(seqid, iprot, oprot)
+ args = read_args(iprot, Get_current_notificationEventId_args)
+ result = Get_current_notificationEventId_result.new()
+ result.success = @handler.get_current_notificationEventId()
+ write_result(result, oprot, 'get_current_notificationEventId', seqid)
+ end
+
end
# HELPER FUNCTIONS AND STRUCTURES
@@ -7974,6 +8018,69 @@ module ThriftHiveMetastore
}
def struct_fields; FIELDS; end
+
+ def validate
+ end
+
+ ::Thrift::Struct.generate_accessors self
+ end
+
+ class Get_next_notification_args
+ include ::Thrift::Struct, ::Thrift::Struct_Union
+ RQST = 1
+
+ FIELDS = {
+ RQST => {:type => ::Thrift::Types::STRUCT, :name => 'rqst', :class => ::NotificationEventRequest}
+ }
+
+ def struct_fields; FIELDS; end
+
+ def validate
+ end
+
+ ::Thrift::Struct.generate_accessors self
+ end
+
+ class Get_next_notification_result
+ include ::Thrift::Struct, ::Thrift::Struct_Union
+ SUCCESS = 0
+
+ FIELDS = {
+ SUCCESS => {:type => ::Thrift::Types::STRUCT, :name => 'success', :class => ::NotificationEventResponse}
+ }
+
+ def struct_fields; FIELDS; end
+
+ def validate
+ end
+
+ ::Thrift::Struct.generate_accessors self
+ end
+
+ class Get_current_notificationEventId_args
+ include ::Thrift::Struct, ::Thrift::Struct_Union
+
+ FIELDS = {
+
+ }
+
+ def struct_fields; FIELDS; end
+
+ def validate
+ end
+
+ ::Thrift::Struct.generate_accessors self
+ end
+
+ class Get_current_notificationEventId_result
+ include ::Thrift::Struct, ::Thrift::Struct_Union
+ SUCCESS = 0
+
+ FIELDS = {
+ SUCCESS => {:type => ::Thrift::Types::STRUCT, :name => 'success', :class => ::CurrentNotificationEventId}
+ }
+
+ def struct_fields; FIELDS; end
def validate
end
Modified: hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java (original)
+++ hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java Fri Jan 23 19:59:11 2015
@@ -77,6 +77,7 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.metastore.api.CommitTxnRequest;
import org.apache.hadoop.hive.metastore.api.CompactionRequest;
import org.apache.hadoop.hive.metastore.api.ConfigValSecurityException;
+import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.DropPartitionsExpr;
import org.apache.hadoop.hive.metastore.api.DropPartitionsRequest;
@@ -111,6 +112,8 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.metastore.api.NoSuchLockException;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
import org.apache.hadoop.hive.metastore.api.NoSuchTxnException;
+import org.apache.hadoop.hive.metastore.api.NotificationEventRequest;
+import org.apache.hadoop.hive.metastore.api.NotificationEventResponse;
import org.apache.hadoop.hive.metastore.api.OpenTxnRequest;
import org.apache.hadoop.hive.metastore.api.OpenTxnsResponse;
import org.apache.hadoop.hive.metastore.api.Partition;
@@ -201,7 +204,9 @@ import org.apache.hadoop.util.Reflection
import org.apache.hadoop.util.StringUtils;
import org.apache.thrift.TException;
import org.apache.thrift.TProcessor;
+import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
+import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.transport.TFramedTransport;
@@ -3120,9 +3125,14 @@ public class HiveMetaStore extends Thrif
oldPart = alterHandler.alterPartition(getMS(), wh, db_name, tbl_name, part_vals, new_part);
+ // Only fetch the table if we actually have a listener
+ Table table = null;
for (MetaStoreEventListener listener : listeners) {
+ if (table == null) {
+ table = getMS().getTable(db_name, tbl_name);
+ }
AlterPartitionEvent alterPartitionEvent =
- new AlterPartitionEvent(oldPart, new_part, true, this);
+ new AlterPartitionEvent(oldPart, new_part, table, true, this);
alterPartitionEvent.setEnvironmentContext(envContext);
listener.onAlterPartition(alterPartitionEvent);
}
@@ -3173,6 +3183,8 @@ public class HiveMetaStore extends Thrif
oldParts = alterHandler.alterPartitions(getMS(), wh, db_name, tbl_name, new_parts);
Iterator<Partition> olditr = oldParts.iterator();
+ // Only fetch the table if we have a listener that needs it.
+ Table table = null;
for (Partition tmpPart : new_parts) {
Partition oldTmpPart = null;
if (olditr.hasNext()) {
@@ -3182,8 +3194,11 @@ public class HiveMetaStore extends Thrif
throw new InvalidOperationException("failed to alterpartitions");
}
for (MetaStoreEventListener listener : listeners) {
+ if (table == null) {
+ table = getMS().getTable(db_name, tbl_name);
+ }
AlterPartitionEvent alterPartitionEvent =
- new AlterPartitionEvent(oldTmpPart, tmpPart, true, this);
+ new AlterPartitionEvent(oldTmpPart, tmpPart, table, true, this);
listener.onAlterPartition(alterPartitionEvent);
}
}
@@ -5362,126 +5377,74 @@ public class HiveMetaStore extends Thrif
// Transaction and locking methods
@Override
public GetOpenTxnsResponse get_open_txns() throws TException {
- try {
- return getTxnHandler().getOpenTxns();
- } catch (MetaException e) {
- throw new TException(e);
- }
+ return getTxnHandler().getOpenTxns();
}
// Transaction and locking methods
@Override
public GetOpenTxnsInfoResponse get_open_txns_info() throws TException {
- try {
- return getTxnHandler().getOpenTxnsInfo();
- } catch (MetaException e) {
- throw new TException(e);
- }
+ return getTxnHandler().getOpenTxnsInfo();
}
@Override
public OpenTxnsResponse open_txns(OpenTxnRequest rqst) throws TException {
- try {
- return getTxnHandler().openTxns(rqst);
- } catch (MetaException e) {
- throw new TException(e);
- }
+ return getTxnHandler().openTxns(rqst);
}
@Override
public void abort_txn(AbortTxnRequest rqst) throws NoSuchTxnException, TException {
- try {
- getTxnHandler().abortTxn(rqst);
- } catch (MetaException e) {
- throw new TException(e);
- }
+ getTxnHandler().abortTxn(rqst);
}
@Override
public void commit_txn(CommitTxnRequest rqst)
throws NoSuchTxnException, TxnAbortedException, TException {
- try {
- getTxnHandler().commitTxn(rqst);
- } catch (MetaException e) {
- throw new TException(e);
- }
+ getTxnHandler().commitTxn(rqst);
}
@Override
public LockResponse lock(LockRequest rqst)
throws NoSuchTxnException, TxnAbortedException, TException {
- try {
- return getTxnHandler().lock(rqst);
- } catch (MetaException e) {
- throw new TException(e);
- }
+ return getTxnHandler().lock(rqst);
}
@Override
public LockResponse check_lock(CheckLockRequest rqst)
throws NoSuchTxnException, TxnAbortedException, NoSuchLockException, TException {
- try {
- return getTxnHandler().checkLock(rqst);
- } catch (MetaException e) {
- throw new TException(e);
- }
+ return getTxnHandler().checkLock(rqst);
}
@Override
public void unlock(UnlockRequest rqst)
throws NoSuchLockException, TxnOpenException, TException {
- try {
- getTxnHandler().unlock(rqst);
- } catch (MetaException e) {
- throw new TException(e);
- }
+ getTxnHandler().unlock(rqst);
}
@Override
public ShowLocksResponse show_locks(ShowLocksRequest rqst) throws TException {
- try {
- return getTxnHandler().showLocks(rqst);
- } catch (MetaException e) {
- throw new TException(e);
- }
+ return getTxnHandler().showLocks(rqst);
}
@Override
public void heartbeat(HeartbeatRequest ids)
throws NoSuchLockException, NoSuchTxnException, TxnAbortedException, TException {
- try {
- getTxnHandler().heartbeat(ids);
- } catch (MetaException e) {
- throw new TException(e);
- }
+ getTxnHandler().heartbeat(ids);
}
@Override
public HeartbeatTxnRangeResponse heartbeat_txn_range(HeartbeatTxnRangeRequest rqst)
throws TException {
- try {
- return getTxnHandler().heartbeatTxnRange(rqst);
- } catch (MetaException e) {
- throw new TException(e);
- }
+ return getTxnHandler().heartbeatTxnRange(rqst);
}
@Override
public void compact(CompactionRequest rqst) throws TException {
- try {
- getTxnHandler().compact(rqst);
- } catch (MetaException e) {
- throw new TException(e);
- }
+ getTxnHandler().compact(rqst);
}
@Override
public ShowCompactResponse show_compact(ShowCompactRequest rqst) throws TException {
- try {
- return getTxnHandler().showCompact(rqst);
- } catch (MetaException e) {
- throw new TException(e);
- }
+ return getTxnHandler().showCompact(rqst);
}
@Override
@@ -5579,6 +5542,19 @@ public class HiveMetaStore extends Thrif
}
return ret;
}
+
+ @Override
+ public NotificationEventResponse get_next_notification(NotificationEventRequest rqst)
+ throws TException {
+ RawStore ms = getMS();
+ return ms.getNextNotification(rqst);
+ }
+
+ @Override
+ public CurrentNotificationEventId get_current_notificationEventId() throws TException {
+ RawStore ms = getMS();
+ return ms.getCurrentNotificationEventId();
+ }
}
@@ -5798,6 +5774,7 @@ public class HiveMetaStore extends Thrif
int maxWorkerThreads = conf.getIntVar(HiveConf.ConfVars.METASTORESERVERMAXTHREADS);
boolean tcpKeepAlive = conf.getBoolVar(HiveConf.ConfVars.METASTORE_TCP_KEEP_ALIVE);
boolean useFramedTransport = conf.getBoolVar(ConfVars.METASTORE_USE_THRIFT_FRAMED_TRANSPORT);
+ boolean useCompactProtocol = conf.getBoolVar(ConfVars.METASTORE_USE_THRIFT_COMPACT_PROTOCOL);
useSasl = conf.getBoolVar(HiveConf.ConfVars.METASTORE_USE_THRIFT_SASL);
TServerTransport serverTransport = tcpKeepAlive ?
@@ -5805,6 +5782,15 @@ public class HiveMetaStore extends Thrif
TProcessor processor;
TTransportFactory transFactory;
+ final TProtocolFactory protocolFactory;
+ final TProtocolFactory inputProtoFactory;
+ if (useCompactProtocol) {
+ protocolFactory = new TCompactProtocol.Factory();
+ inputProtoFactory = new TCompactProtocol.Factory(maxMessageSize, maxMessageSize);
+ } else {
+ protocolFactory = new TBinaryProtocol.Factory();
+ inputProtoFactory = new TBinaryProtocol.Factory(true, true, maxMessageSize, maxMessageSize);
+ }
HMSHandler baseHandler = new HiveMetaStore.HMSHandler("new db based metaserver", conf,
false);
IHMSHandler handler = newRetryingHMSHandler(baseHandler, conf);
@@ -5844,9 +5830,8 @@ public class HiveMetaStore extends Thrif
TThreadPoolServer.Args args = new TThreadPoolServer.Args(serverTransport)
.processor(processor)
.transportFactory(transFactory)
- .protocolFactory(new TCompactProtocol.Factory())
- .inputProtocolFactory(
- new TCompactProtocol.Factory(maxMessageSize, maxMessageSize))
+ .protocolFactory(protocolFactory)
+ .inputProtocolFactory(inputProtoFactory)
.minWorkerThreads(minWorkerThreads)
.maxWorkerThreads(maxWorkerThreads);
Modified: hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java (original)
+++ hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java Fri Jan 23 19:59:11 2015
@@ -62,6 +62,7 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.metastore.api.CompactionRequest;
import org.apache.hadoop.hive.metastore.api.CompactionType;
import org.apache.hadoop.hive.metastore.api.ConfigValSecurityException;
+import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.DropPartitionsExpr;
import org.apache.hadoop.hive.metastore.api.DropPartitionsRequest;
@@ -94,6 +95,9 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.metastore.api.NoSuchLockException;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
import org.apache.hadoop.hive.metastore.api.NoSuchTxnException;
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.api.NotificationEventRequest;
+import org.apache.hadoop.hive.metastore.api.NotificationEventResponse;
import org.apache.hadoop.hive.metastore.api.OpenTxnRequest;
import org.apache.hadoop.hive.metastore.api.OpenTxnsResponse;
import org.apache.hadoop.hive.metastore.api.Partition;
@@ -131,7 +135,9 @@ import org.apache.hadoop.security.UserGr
import org.apache.hadoop.util.StringUtils;
import org.apache.thrift.TApplicationException;
import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
+import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
@@ -345,6 +351,7 @@ public class HiveMetaStoreClient impleme
TTransportException tte = null;
boolean useSasl = conf.getBoolVar(ConfVars.METASTORE_USE_THRIFT_SASL);
boolean useFramedTransport = conf.getBoolVar(ConfVars.METASTORE_USE_THRIFT_FRAMED_TRANSPORT);
+ boolean useCompactProtocol = conf.getBoolVar(ConfVars.METASTORE_USE_THRIFT_COMPACT_PROTOCOL);
int clientSocketTimeout = (int) conf.getTimeVar(
ConfVars.METASTORE_CLIENT_SOCKET_TIMEOUT, TimeUnit.MILLISECONDS);
@@ -386,8 +393,13 @@ public class HiveMetaStoreClient impleme
} else if (useFramedTransport) {
transport = new TFramedTransport(transport);
}
-
- client = new ThriftHiveMetastore.Client(new TCompactProtocol(transport));
+ final TProtocol protocol;
+ if (useCompactProtocol) {
+ protocol = new TCompactProtocol(transport);
+ } else {
+ protocol = new TBinaryProtocol(transport);
+ }
+ client = new ThriftHiveMetastore.Client(protocol);
try {
transport.open();
isConnected = true;
@@ -1421,7 +1433,8 @@ public class HiveMetaStoreClient impleme
@Override
public Partition getPartition(String db, String tableName, String partName)
throws MetaException, TException, UnknownTableException, NoSuchObjectException {
- return deepCopy(filterHook.filterPartition(client.get_partition_by_name(db, tableName, partName)));
+ return deepCopy(
+ filterHook.filterPartition(client.get_partition_by_name(db, tableName, partName)));
}
public Partition appendPartitionByName(String dbName, String tableName, String partName)
@@ -1829,6 +1842,31 @@ public class HiveMetaStoreClient impleme
return client.show_compact(new ShowCompactRequest());
}
+ @Override
+ public NotificationEventResponse getNextNotification(long lastEventId, int maxEvents,
+ NotificationFilter filter) throws TException {
+ NotificationEventRequest rqst = new NotificationEventRequest(lastEventId);
+ rqst.setMaxEvents(maxEvents);
+ NotificationEventResponse rsp = client.get_next_notification(rqst);
+ LOG.debug("Got back " + rsp.getEventsSize() + " events");
+ if (filter == null) {
+ return rsp;
+ } else {
+ NotificationEventResponse filtered = new NotificationEventResponse();
+ if (rsp != null && rsp.getEvents() != null) {
+ for (NotificationEvent e : rsp.getEvents()) {
+ if (filter.accept(e)) filtered.addToEvents(e);
+ }
+ }
+ return filtered;
+ }
+ }
+
+ @Override
+ public CurrentNotificationEventId getCurrentNotificationEventId() throws TException {
+ return client.get_current_notificationEventId();
+ }
+
/**
* Creates a synchronized wrapper for any {@link IMetaStoreClient}.
* This may be used by multi-threaded applications until we have
Modified: hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java (original)
+++ hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java Fri Jan 23 19:59:11 2015
@@ -21,12 +21,16 @@ package org.apache.hadoop.hive.metastore
import org.apache.hadoop.hive.common.ValidTxnList;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.CompactionType;
+import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
import org.apache.hadoop.hive.metastore.api.GetOpenTxnsInfoResponse;
import org.apache.hadoop.hive.metastore.api.HeartbeatTxnRangeResponse;
import org.apache.hadoop.hive.metastore.api.LockRequest;
import org.apache.hadoop.hive.metastore.api.LockResponse;
import org.apache.hadoop.hive.metastore.api.NoSuchLockException;
import org.apache.hadoop.hive.metastore.api.NoSuchTxnException;
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.api.NotificationEventRequest;
+import org.apache.hadoop.hive.metastore.api.NotificationEventResponse;
import org.apache.hadoop.hive.metastore.api.OpenTxnsResponse;
import org.apache.hadoop.hive.metastore.api.PartitionSpec;
import org.apache.hadoop.hive.metastore.api.ShowCompactResponse;
@@ -1306,6 +1310,41 @@ public interface IMetaStoreClient {
*/
ShowCompactResponse showCompactions() throws TException;
+ /**
+ * A filter provided by the client that determines if a given notification event should be
+ * returned.
+ */
+ interface NotificationFilter {
+ /**
+ * Whether a notification event should be accepted
+ * @param event
+ * @return if true, event will be added to list, if false it will be ignored
+ */
+ boolean accept(NotificationEvent event);
+ }
+
+ /**
+ * Get the next set of notifications from the database.
+ * @param lastEventId The last event id that was consumed by this reader. The returned
+ * notifications will start at the next eventId available after this eventId.
+ * @param maxEvents Maximum number of events to return. If < 1, then all available events will
+ * be returned.
+ * @param filter User provided filter to remove unwanted events. If null, all events will be
+ * returned.
+ * @return list of notifications, sorted by eventId. It is guaranteed that the events are in
+ * the order that the operations were done on the database.
+ * @throws TException
+ */
+ NotificationEventResponse getNextNotification(long lastEventId, int maxEvents,
+ NotificationFilter filter) throws TException;
+
+ /**
+ * Get the last used notification event id.
+ * @return last used id
+ * @throws TException
+ */
+ CurrentNotificationEventId getCurrentNotificationEventId() throws TException;
+
class IncompatibleMetastoreException extends MetaException {
IncompatibleMetastoreException(String message) {
super(message);
Modified: hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java (original)
+++ hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java Fri Jan 23 19:59:11 2015
@@ -67,6 +67,7 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsDesc;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
+import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Function;
@@ -80,6 +81,9 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.metastore.api.InvalidPartitionException;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.api.NotificationEventRequest;
+import org.apache.hadoop.hive.metastore.api.NotificationEventResponse;
import org.apache.hadoop.hive.metastore.api.Order;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.PartitionEventType;
@@ -107,6 +111,8 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.metastore.model.MGlobalPrivilege;
import org.apache.hadoop.hive.metastore.model.MIndex;
import org.apache.hadoop.hive.metastore.model.MMasterKey;
+import org.apache.hadoop.hive.metastore.model.MNotificationLog;
+import org.apache.hadoop.hive.metastore.model.MNotificationNextId;
import org.apache.hadoop.hive.metastore.model.MOrder;
import org.apache.hadoop.hive.metastore.model.MPartition;
import org.apache.hadoop.hive.metastore.model.MPartitionColumnPrivilege;
@@ -6962,4 +6968,128 @@ public class ObjectStore implements RawS
}
return funcs;
}
+
+ @Override
+ public NotificationEventResponse getNextNotification(NotificationEventRequest rqst) {
+ boolean commited = false;
+ try {
+ openTransaction();
+ long lastEvent = rqst.getLastEvent();
+ Query query = pm.newQuery(MNotificationLog.class, "eventId > lastEvent");
+ query.declareParameters("java.lang.Long lastEvent");
+ query.setOrdering("eventId ascending");
+ Collection<MNotificationLog> events = (Collection)query.execute(lastEvent);
+ commited = commitTransaction();
+ if (events == null) {
+ return null;
+ }
+ Iterator<MNotificationLog> i = events.iterator();
+ NotificationEventResponse result = new NotificationEventResponse();
+ int maxEvents = rqst.getMaxEvents() > 0 ? rqst.getMaxEvents() : Integer.MAX_VALUE;
+ int numEvents = 0;
+ while (i.hasNext() && numEvents++ < maxEvents) {
+ result.addToEvents(translateDbToThrift(i.next()));
+ }
+ return result;
+ } finally {
+ if (!commited) {
+ rollbackTransaction();
+ return null;
+ }
+ }
+ }
+
+ @Override
+ public void addNotificationEvent(NotificationEvent entry) {
+ boolean commited = false;
+ try {
+ openTransaction();
+ Query query = pm.newQuery(MNotificationNextId.class);
+ Collection<MNotificationNextId> ids = (Collection) query.execute();
+ MNotificationNextId id = null;
+ boolean needToPersistId;
+ if (ids == null || ids.size() == 0) {
+ id = new MNotificationNextId(1L);
+ needToPersistId = true;
+ } else {
+ id = ids.iterator().next();
+ needToPersistId = false;
+ }
+ entry.setEventId(id.getNextEventId());
+ id.incrementEventId();
+ if (needToPersistId) pm.makePersistent(id);
+ pm.makePersistent(translateThriftToDb(entry));
+ commited = commitTransaction();
+ } finally {
+ if (!commited) {
+ rollbackTransaction();
+ }
+ }
+ }
+
+ @Override
+ public void cleanNotificationEvents(int olderThan) {
+ boolean commited = false;
+ try {
+ openTransaction();
+ long tmp = System.currentTimeMillis() / 1000 - olderThan;
+ int tooOld = (tmp > Integer.MAX_VALUE) ? 0 : (int)tmp;
+ Query query = pm.newQuery(MNotificationLog.class, "eventTime < tooOld");
+ query.declareParameters("java.lang.Integer tooOld");
+ Collection<MNotificationLog> toBeRemoved = (Collection)query.execute(tooOld);
+ if (toBeRemoved != null && toBeRemoved.size() > 0) {
+ pm.deletePersistent(toBeRemoved);
+ }
+ commited = commitTransaction();
+ } finally {
+ if (!commited) {
+ rollbackTransaction();
+ }
+ }
+ }
+
+ @Override
+ public CurrentNotificationEventId getCurrentNotificationEventId() {
+ boolean commited = false;
+ try {
+ openTransaction();
+ Query query = pm.newQuery(MNotificationNextId.class);
+ Collection<MNotificationNextId> ids = (Collection)query.execute();
+ long id = 0;
+ if (ids != null && ids.size() > 0) {
+ id = ids.iterator().next().getNextEventId() - 1;
+ }
+ commited = commitTransaction();
+ return new CurrentNotificationEventId(id);
+ } finally {
+ if (!commited) {
+ rollbackTransaction();
+ }
+ }
+ }
+
+ private MNotificationLog translateThriftToDb(NotificationEvent entry) {
+ MNotificationLog dbEntry = new MNotificationLog();
+ dbEntry.setEventId(entry.getEventId());
+ dbEntry.setEventTime(entry.getEventTime());
+ dbEntry.setEventType(entry.getEventType());
+ dbEntry.setDbName(entry.getDbName());
+ dbEntry.setTableName(entry.getTableName());
+ dbEntry.setMessage(entry.getMessage());
+ return dbEntry;
+ }
+
+ private NotificationEvent translateDbToThrift(MNotificationLog dbEvent) {
+ NotificationEvent event = new NotificationEvent();
+ event.setEventId(dbEvent.getEventId());
+ event.setEventTime(dbEvent.getEventTime());
+ event.setEventType(dbEvent.getEventType());
+ event.setDbName(dbEvent.getDbName());
+ event.setTableName(dbEvent.getTableName());
+ event.setMessage((dbEvent.getMessage()));
+ return event;
+ }
+
+
+
}
Modified: hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/RawStore.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/RawStore.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/RawStore.java (original)
+++ hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/RawStore.java Fri Jan 23 19:59:11 2015
@@ -24,11 +24,13 @@ import java.lang.annotation.RetentionPol
import java.lang.annotation.Target;
import java.util.List;
import java.util.Map;
+import java.util.SortedSet;
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.hive.metastore.api.AggrStats;
import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
+import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.Function;
import org.apache.hadoop.hive.metastore.api.HiveObjectPrivilege;
@@ -38,6 +40,9 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.metastore.api.InvalidPartitionException;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.api.NotificationEventRequest;
+import org.apache.hadoop.hive.metastore.api.NotificationEventResponse;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.PartitionEventType;
import org.apache.hadoop.hive.metastore.api.PrincipalPrivilegeSet;
@@ -556,5 +561,33 @@ public interface RawStore extends Config
public AggrStats get_aggr_stats_for(String dbName, String tblName,
List<String> partNames, List<String> colNames) throws MetaException, NoSuchObjectException;
+
+ /**
+ * Get the next notification event.
+ * @param rqst Request containing information on the last processed notification.
+ * @return list of notifications, sorted by eventId
+ */
+ public NotificationEventResponse getNextNotification(NotificationEventRequest rqst);
+
+
+ /**
+ * Add a notification entry. This should only be called from inside the metastore
+ * @param event the notification to add
+ */
+ public void addNotificationEvent(NotificationEvent event);
+
+ /**
+ * Remove older notification events.
+ * @param olderThan Remove any events older than a given number of seconds
+ */
+ public void cleanNotificationEvents(int olderThan);
+
+ /**
+ * Get the last issued notification event id. This is intended for use by the export command
+ * so that users can determine the state of the system at the point of the export,
+ * and determine which notification events happened before or after the export.
+ * @return
+ */
+ public CurrentNotificationEventId getCurrentNotificationEventId();
}
Modified: hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/events/AlterPartitionEvent.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/events/AlterPartitionEvent.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/events/AlterPartitionEvent.java (original)
+++ hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/events/AlterPartitionEvent.java Fri Jan 23 19:59:11 2015
@@ -20,17 +20,20 @@ package org.apache.hadoop.hive.metastore
import org.apache.hadoop.hive.metastore.HiveMetaStore.HMSHandler;
import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
public class AlterPartitionEvent extends ListenerEvent {
private final Partition oldPart;
private final Partition newPart;
+ private final Table table;
- public AlterPartitionEvent(Partition oldPart, Partition newPart,
+ public AlterPartitionEvent(Partition oldPart, Partition newPart, Table table,
boolean status, HMSHandler handler) {
super(status, handler);
this.oldPart = oldPart;
this.newPart = newPart;
+ this.table = table;
}
/**
@@ -47,4 +50,12 @@ public class AlterPartitionEvent extends
public Partition getNewPartition() {
return newPart;
}
+
+ /**
+ * Get the table this partition is in
+ * @return
+ */
+ public Table getTable() {
+ return table;
+ }
}
Modified: hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java (original)
+++ hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java Fri Jan 23 19:59:11 2015
@@ -52,51 +52,58 @@ public class CompactionTxnHandler extend
* or runAs set since these are only potential compactions not actual ones.
*/
public Set<CompactionInfo> findPotentialCompactions(int maxAborted) throws MetaException {
- Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
+ Connection dbConn = null;
Set<CompactionInfo> response = new HashSet<CompactionInfo>();
Statement stmt = null;
try {
- stmt = dbConn.createStatement();
- // Check for completed transactions
- String s = "select distinct ctc_database, ctc_table, " +
+ try {
+ dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
+ stmt = dbConn.createStatement();
+ // Check for completed transactions
+ String s = "select distinct ctc_database, ctc_table, " +
"ctc_partition from COMPLETED_TXN_COMPONENTS";
- LOG.debug("Going to execute query <" + s + ">");
- ResultSet rs = stmt.executeQuery(s);
- while (rs.next()) {
- CompactionInfo info = new CompactionInfo();
- info.dbname = rs.getString(1);
- info.tableName = rs.getString(2);
- info.partName = rs.getString(3);
- response.add(info);
- }
+ LOG.debug("Going to execute query <" + s + ">");
+ ResultSet rs = stmt.executeQuery(s);
+ while (rs.next()) {
+ CompactionInfo info = new CompactionInfo();
+ info.dbname = rs.getString(1);
+ info.tableName = rs.getString(2);
+ info.partName = rs.getString(3);
+ response.add(info);
+ }
- // Check for aborted txns
- s = "select tc_database, tc_table, tc_partition " +
+ // Check for aborted txns
+ s = "select tc_database, tc_table, tc_partition " +
"from TXNS, TXN_COMPONENTS " +
"where txn_id = tc_txnid and txn_state = '" + TXN_ABORTED + "' " +
"group by tc_database, tc_table, tc_partition " +
"having count(*) > " + maxAborted;
- LOG.debug("Going to execute query <" + s + ">");
- rs = stmt.executeQuery(s);
- while (rs.next()) {
- CompactionInfo info = new CompactionInfo();
- info.dbname = rs.getString(1);
- info.tableName = rs.getString(2);
- info.partName = rs.getString(3);
- info.tooManyAborts = true;
- response.add(info);
- }
+ LOG.debug("Going to execute query <" + s + ">");
+ rs = stmt.executeQuery(s);
+ while (rs.next()) {
+ CompactionInfo info = new CompactionInfo();
+ info.dbname = rs.getString(1);
+ info.tableName = rs.getString(2);
+ info.partName = rs.getString(3);
+ info.tooManyAborts = true;
+ response.add(info);
+ }
- LOG.debug("Going to rollback");
- dbConn.rollback();
- } catch (SQLException e) {
- LOG.error("Unable to connect to transaction database " + e.getMessage());
- } finally {
- closeDbConn(dbConn);
- closeStmt(stmt);
+ LOG.debug("Going to rollback");
+ dbConn.rollback();
+ } catch (SQLException e) {
+ LOG.error("Unable to connect to transaction database " + e.getMessage());
+ checkRetryable(dbConn, e, "findPotentialCompactions");
+ } finally {
+ closeDbConn(dbConn);
+ closeStmt(stmt);
+ }
+ return response;
+ }
+ catch (RetryException e) {
+ return findPotentialCompactions(maxAborted);
}
- return response;
}
/**
@@ -107,35 +114,31 @@ public class CompactionTxnHandler extend
*/
public void setRunAs(long cq_id, String user) throws MetaException {
try {
- Connection dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
+ Connection dbConn = null;
Statement stmt = null;
try {
- stmt = dbConn.createStatement();
- String s = "update COMPACTION_QUEUE set cq_run_as = '" + user + "' where cq_id = " + cq_id;
- LOG.debug("Going to execute update <" + s + ">");
- if (stmt.executeUpdate(s) != 1) {
- LOG.error("Unable to update compaction record");
- LOG.debug("Going to rollback");
- dbConn.rollback();
- }
- LOG.debug("Going to commit");
- dbConn.commit();
- } catch (SQLException e) {
- LOG.error("Unable to update compaction queue, " + e.getMessage());
- try {
- LOG.debug("Going to rollback");
- dbConn.rollback();
- } catch (SQLException e1) {
- }
- detectDeadlock(dbConn, e, "setRunAs");
- } finally {
- closeDbConn(dbConn);
- closeStmt(stmt);
- }
- } catch (DeadlockException e) {
+ dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
+ stmt = dbConn.createStatement();
+ String s = "update COMPACTION_QUEUE set cq_run_as = '" + user + "' where cq_id = " + cq_id;
+ LOG.debug("Going to execute update <" + s + ">");
+ if (stmt.executeUpdate(s) != 1) {
+ LOG.error("Unable to update compaction record");
+ LOG.debug("Going to rollback");
+ dbConn.rollback();
+ }
+ LOG.debug("Going to commit");
+ dbConn.commit();
+ } catch (SQLException e) {
+ LOG.error("Unable to update compaction queue, " + e.getMessage());
+ LOG.debug("Going to rollback");
+ rollbackDBConn(dbConn);
+ checkRetryable(dbConn, e, "setRunAs");
+ } finally {
+ closeDbConn(dbConn);
+ closeStmt(stmt);
+ }
+ } catch (RetryException e) {
setRunAs(cq_id, user);
- } finally {
- deadlockCnt = 0;
}
}
@@ -147,14 +150,15 @@ public class CompactionTxnHandler extend
*/
public CompactionInfo findNextToCompact(String workerId) throws MetaException {
try {
- Connection dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
+ Connection dbConn = null;
CompactionInfo info = new CompactionInfo();
Statement stmt = null;
try {
+ dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
stmt = dbConn.createStatement();
String s = "select cq_id, cq_database, cq_table, cq_partition, " +
- "cq_type from COMPACTION_QUEUE where cq_state = '" + INITIATED_STATE + "'";
+ "cq_type from COMPACTION_QUEUE where cq_state = '" + INITIATED_STATE + "'";
LOG.debug("Going to execute query <" + s + ">");
ResultSet rs = stmt.executeQuery(s);
if (!rs.next()) {
@@ -175,7 +179,7 @@ public class CompactionTxnHandler extend
// Now, update this record as being worked on by this worker.
long now = getDbTime(dbConn);
s = "update COMPACTION_QUEUE set cq_worker_id = '" + workerId + "', " +
- "cq_start = " + now + ", cq_state = '" + WORKING_STATE + "' where cq_id = " + info.id;
+ "cq_start = " + now + ", cq_state = '" + WORKING_STATE + "' where cq_id = " + info.id;
LOG.debug("Going to execute update <" + s + ">");
if (stmt.executeUpdate(s) != 1) {
LOG.error("Unable to update compaction record");
@@ -187,38 +191,34 @@ public class CompactionTxnHandler extend
return info;
} catch (SQLException e) {
LOG.error("Unable to select next element for compaction, " + e.getMessage());
- try {
- LOG.debug("Going to rollback");
- dbConn.rollback();
- } catch (SQLException e1) {
- }
- detectDeadlock(dbConn, e, "findNextToCompact");
+ LOG.debug("Going to rollback");
+ rollbackDBConn(dbConn);
+ checkRetryable(dbConn, e, "findNextToCompact");
throw new MetaException("Unable to connect to transaction database " +
- StringUtils.stringifyException(e));
+ StringUtils.stringifyException(e));
} finally {
closeDbConn(dbConn);
closeStmt(stmt);
}
- } catch (DeadlockException e) {
+ } catch (RetryException e) {
return findNextToCompact(workerId);
- } finally {
- deadlockCnt = 0;
}
}
/**
* This will mark an entry in the queue as compacted
* and put it in the ready to clean state.
- * @param info info on the compaciton entry to mark as compacted.
+ * @param info info on the compaction entry to mark as compacted.
*/
public void markCompacted(CompactionInfo info) throws MetaException {
try {
- Connection dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
+ Connection dbConn = null;
Statement stmt = null;
try {
+ dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
stmt = dbConn.createStatement();
String s = "update COMPACTION_QUEUE set cq_state = '" + READY_FOR_CLEANING + "', " +
- "cq_worker_id = null where cq_id = " + info.id;
+ "cq_worker_id = null where cq_id = " + info.id;
LOG.debug("Going to execute update <" + s + ">");
if (stmt.executeUpdate(s) != 1) {
LOG.error("Unable to update compaction record");
@@ -228,23 +228,18 @@ public class CompactionTxnHandler extend
LOG.debug("Going to commit");
dbConn.commit();
} catch (SQLException e) {
- try {
- LOG.error("Unable to update compaction queue " + e.getMessage());
- LOG.debug("Going to rollback");
- dbConn.rollback();
- } catch (SQLException e1) {
- }
- detectDeadlock(dbConn, e, "markCompacted");
+ LOG.error("Unable to update compaction queue " + e.getMessage());
+ LOG.debug("Going to rollback");
+ rollbackDBConn(dbConn);
+ checkRetryable(dbConn, e, "markCompacted");
throw new MetaException("Unable to connect to transaction database " +
- StringUtils.stringifyException(e));
+ StringUtils.stringifyException(e));
} finally {
closeDbConn(dbConn);
closeStmt(stmt);
}
- } catch (DeadlockException e) {
+ } catch (RetryException e) {
markCompacted(info);
- } finally {
- deadlockCnt = 0;
}
}
@@ -254,45 +249,48 @@ public class CompactionTxnHandler extend
* @return information on the entry in the queue.
*/
public List<CompactionInfo> findReadyToClean() throws MetaException {
- Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
+ Connection dbConn = null;
List<CompactionInfo> rc = new ArrayList<CompactionInfo>();
Statement stmt = null;
try {
- stmt = dbConn.createStatement();
- String s = "select cq_id, cq_database, cq_table, cq_partition, " +
+ try {
+ dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
+ stmt = dbConn.createStatement();
+ String s = "select cq_id, cq_database, cq_table, cq_partition, " +
"cq_type, cq_run_as from COMPACTION_QUEUE where cq_state = '" + READY_FOR_CLEANING + "'";
- LOG.debug("Going to execute query <" + s + ">");
- ResultSet rs = stmt.executeQuery(s);
- while (rs.next()) {
- CompactionInfo info = new CompactionInfo();
- info.id = rs.getLong(1);
- info.dbname = rs.getString(2);
- info.tableName = rs.getString(3);
- info.partName = rs.getString(4);
- switch (rs.getString(5).charAt(0)) {
- case MAJOR_TYPE: info.type = CompactionType.MAJOR; break;
- case MINOR_TYPE: info.type = CompactionType.MINOR; break;
- default: throw new MetaException("Unexpected compaction type " + rs.getString(5));
+ LOG.debug("Going to execute query <" + s + ">");
+ ResultSet rs = stmt.executeQuery(s);
+ while (rs.next()) {
+ CompactionInfo info = new CompactionInfo();
+ info.id = rs.getLong(1);
+ info.dbname = rs.getString(2);
+ info.tableName = rs.getString(3);
+ info.partName = rs.getString(4);
+ switch (rs.getString(5).charAt(0)) {
+ case MAJOR_TYPE: info.type = CompactionType.MAJOR; break;
+ case MINOR_TYPE: info.type = CompactionType.MINOR; break;
+ default: throw new MetaException("Unexpected compaction type " + rs.getString(5));
+ }
+ info.runAs = rs.getString(6);
+ rc.add(info);
}
- info.runAs = rs.getString(6);
- rc.add(info);
- }
- LOG.debug("Going to rollback");
- dbConn.rollback();
- return rc;
- } catch (SQLException e) {
- LOG.error("Unable to select next element for cleaning, " + e.getMessage());
- try {
LOG.debug("Going to rollback");
dbConn.rollback();
- } catch (SQLException e1) {
- }
- throw new MetaException("Unable to connect to transaction database " +
+ return rc;
+ } catch (SQLException e) {
+ LOG.error("Unable to select next element for cleaning, " + e.getMessage());
+ LOG.debug("Going to rollback");
+ rollbackDBConn(dbConn);
+ checkRetryable(dbConn, e, "findReadyToClean");
+ throw new MetaException("Unable to connect to transaction database " +
StringUtils.stringifyException(e));
- } finally {
- closeDbConn(dbConn);
- closeStmt(stmt);
+ } finally {
+ closeDbConn(dbConn);
+ closeStmt(stmt);
+ }
+ } catch (RetryException e) {
+ return findReadyToClean();
}
}
@@ -303,9 +301,10 @@ public class CompactionTxnHandler extend
*/
public void markCleaned(CompactionInfo info) throws MetaException {
try {
- Connection dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
+ Connection dbConn = null;
Statement stmt = null;
try {
+ dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
stmt = dbConn.createStatement();
String s = "delete from COMPACTION_QUEUE where cq_id = " + info.id;
LOG.debug("Going to execute update <" + s + ">");
@@ -318,20 +317,20 @@ public class CompactionTxnHandler extend
// Remove entries from completed_txn_components as well, so we don't start looking there
// again.
s = "delete from COMPLETED_TXN_COMPONENTS where ctc_database = '" + info.dbname + "' and " +
- "ctc_table = '" + info.tableName + "'";
+ "ctc_table = '" + info.tableName + "'";
if (info.partName != null) {
s += " and ctc_partition = '" + info.partName + "'";
}
LOG.debug("Going to execute update <" + s + ">");
if (stmt.executeUpdate(s) < 1) {
LOG.error("Expected to remove at least one row from completed_txn_components when " +
- "marking compaction entry as clean!");
+ "marking compaction entry as clean!");
}
s = "select txn_id from TXNS, TXN_COMPONENTS where txn_id = tc_txnid and txn_state = '" +
- TXN_ABORTED + "' and tc_database = '" + info.dbname + "' and tc_table = '" +
- info.tableName + "'";
+ TXN_ABORTED + "' and tc_database = '" + info.dbname + "' and tc_table = '" +
+ info.tableName + "'";
if (info.partName != null) s += " and tc_partition = '" + info.partName + "'";
LOG.debug("Going to execute update <" + s + ">");
ResultSet rs = stmt.executeQuery(s);
@@ -371,23 +370,18 @@ public class CompactionTxnHandler extend
LOG.debug("Going to commit");
dbConn.commit();
} catch (SQLException e) {
- try {
- LOG.error("Unable to delete from compaction queue " + e.getMessage());
- LOG.debug("Going to rollback");
- dbConn.rollback();
- } catch (SQLException e1) {
- }
- detectDeadlock(dbConn, e, "markCleaned");
+ LOG.error("Unable to delete from compaction queue " + e.getMessage());
+ LOG.debug("Going to rollback");
+ rollbackDBConn(dbConn);
+ checkRetryable(dbConn, e, "markCleaned");
throw new MetaException("Unable to connect to transaction database " +
- StringUtils.stringifyException(e));
+ StringUtils.stringifyException(e));
} finally {
closeDbConn(dbConn);
closeStmt(stmt);
}
- } catch (DeadlockException e) {
+ } catch (RetryException e) {
markCleaned(info);
- } finally {
- deadlockCnt = 0;
}
}
@@ -396,13 +390,14 @@ public class CompactionTxnHandler extend
*/
public void cleanEmptyAbortedTxns() throws MetaException {
try {
- Connection dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
+ Connection dbConn = null;
Statement stmt = null;
try {
+ dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
stmt = dbConn.createStatement();
String s = "select txn_id from TXNS where " +
- "txn_id not in (select tc_txnid from TXN_COMPONENTS) and " +
- "txn_state = '" + TXN_ABORTED + "'";
+ "txn_id not in (select tc_txnid from TXN_COMPONENTS) and " +
+ "txn_state = '" + TXN_ABORTED + "'";
LOG.debug("Going to execute query <" + s + ">");
ResultSet rs = stmt.executeQuery(s);
Set<Long> txnids = new HashSet<Long>();
@@ -425,21 +420,16 @@ public class CompactionTxnHandler extend
} catch (SQLException e) {
LOG.error("Unable to delete from txns table " + e.getMessage());
LOG.debug("Going to rollback");
- try {
- dbConn.rollback();
- } catch (SQLException e1) {
- }
- detectDeadlock(dbConn, e, "cleanEmptyAbortedTxns");
+ rollbackDBConn(dbConn);
+ checkRetryable(dbConn, e, "cleanEmptyAbortedTxns");
throw new MetaException("Unable to connect to transaction database " +
- StringUtils.stringifyException(e));
+ StringUtils.stringifyException(e));
} finally {
closeDbConn(dbConn);
closeStmt(stmt);
}
- } catch (DeadlockException e) {
+ } catch (RetryException e) {
cleanEmptyAbortedTxns();
- } finally {
- deadlockCnt = 0;
}
}
@@ -454,13 +444,14 @@ public class CompactionTxnHandler extend
*/
public void revokeFromLocalWorkers(String hostname) throws MetaException {
try {
- Connection dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
+ Connection dbConn = null;
Statement stmt = null;
try {
+ dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
stmt = dbConn.createStatement();
String s = "update COMPACTION_QUEUE set cq_worker_id = null, cq_start = null, cq_state = '"
- + INITIATED_STATE+ "' where cq_state = '" + WORKING_STATE + "' and cq_worker_id like '"
- + hostname + "%'";
+ + INITIATED_STATE+ "' where cq_state = '" + WORKING_STATE + "' and cq_worker_id like '"
+ + hostname + "%'";
LOG.debug("Going to execute update <" + s + ">");
// It isn't an error if the following returns no rows, as the local workers could have died
// with nothing assigned to them.
@@ -468,24 +459,19 @@ public class CompactionTxnHandler extend
LOG.debug("Going to commit");
dbConn.commit();
} catch (SQLException e) {
- try {
- LOG.error("Unable to change dead worker's records back to initiated state " +
- e.getMessage());
- LOG.debug("Going to rollback");
- dbConn.rollback();
- } catch (SQLException e1) {
- }
- detectDeadlock(dbConn, e, "revokeFromLocalWorkers");
+ LOG.error("Unable to change dead worker's records back to initiated state " +
+ e.getMessage());
+ LOG.debug("Going to rollback");
+ rollbackDBConn(dbConn);
+ checkRetryable(dbConn, e, "revokeFromLocalWorkers");
throw new MetaException("Unable to connect to transaction database " +
- StringUtils.stringifyException(e));
+ StringUtils.stringifyException(e));
} finally {
closeDbConn(dbConn);
closeStmt(stmt);
}
- } catch (DeadlockException e) {
+ } catch (RetryException e) {
revokeFromLocalWorkers(hostname);
- } finally {
- deadlockCnt = 0;
}
}
@@ -500,14 +486,15 @@ public class CompactionTxnHandler extend
*/
public void revokeTimedoutWorkers(long timeout) throws MetaException {
try {
- Connection dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
- long latestValidStart = getDbTime(dbConn) - timeout;
+ Connection dbConn = null;
Statement stmt = null;
try {
+ dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
+ long latestValidStart = getDbTime(dbConn) - timeout;
stmt = dbConn.createStatement();
String s = "update COMPACTION_QUEUE set cq_worker_id = null, cq_start = null, cq_state = '"
- + INITIATED_STATE+ "' where cq_state = '" + WORKING_STATE + "' and cq_start < "
- + latestValidStart;
+ + INITIATED_STATE+ "' where cq_state = '" + WORKING_STATE + "' and cq_start < "
+ + latestValidStart;
LOG.debug("Going to execute update <" + s + ">");
// It isn't an error if the following returns no rows, as the local workers could have died
// with nothing assigned to them.
@@ -515,24 +502,19 @@ public class CompactionTxnHandler extend
LOG.debug("Going to commit");
dbConn.commit();
} catch (SQLException e) {
- try {
- LOG.error("Unable to change dead worker's records back to initiated state " +
- e.getMessage());
- LOG.debug("Going to rollback");
- dbConn.rollback();
- } catch (SQLException e1) {
- }
- detectDeadlock(dbConn, e, "revokeTimedoutWorkers");
+ LOG.error("Unable to change dead worker's records back to initiated state " +
+ e.getMessage());
+ LOG.debug("Going to rollback");
+ rollbackDBConn(dbConn);
+ checkRetryable(dbConn, e, "revokeTimedoutWorkers");
throw new MetaException("Unable to connect to transaction database " +
- StringUtils.stringifyException(e));
+ StringUtils.stringifyException(e));
} finally {
closeDbConn(dbConn);
closeStmt(stmt);
}
- } catch (DeadlockException e) {
+ } catch (RetryException e) {
revokeTimedoutWorkers(timeout);
- } finally {
- deadlockCnt = 0;
}
}
@@ -543,53 +525,55 @@ public class CompactionTxnHandler extend
* @throws MetaException
*/
public List<String> findColumnsWithStats(CompactionInfo ci) throws MetaException {
- Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
+ Connection dbConn = null;
Statement stmt = null;
ResultSet rs = null;
try {
- String quote = getIdentifierQuoteString(dbConn);
- stmt = dbConn.createStatement();
- StringBuilder bldr = new StringBuilder();
- bldr.append("SELECT ").append(quote).append("COLUMN_NAME").append(quote)
+ try {
+ dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
+ String quote = getIdentifierQuoteString(dbConn);
+ stmt = dbConn.createStatement();
+ StringBuilder bldr = new StringBuilder();
+ bldr.append("SELECT ").append(quote).append("COLUMN_NAME").append(quote)
.append(" FROM ")
.append(quote).append((ci.partName == null ? "TAB_COL_STATS" : "PART_COL_STATS"))
- .append(quote)
+ .append(quote)
.append(" WHERE ")
.append(quote).append("DB_NAME").append(quote).append(" = '").append(ci.dbname)
- .append("' AND ").append(quote).append("TABLE_NAME").append(quote)
- .append(" = '").append(ci.tableName).append("'");
- if (ci.partName != null) {
- bldr.append(" AND ").append(quote).append("PARTITION_NAME").append(quote).append(" = '")
+ .append("' AND ").append(quote).append("TABLE_NAME").append(quote)
+ .append(" = '").append(ci.tableName).append("'");
+ if (ci.partName != null) {
+ bldr.append(" AND ").append(quote).append("PARTITION_NAME").append(quote).append(" = '")
.append(ci.partName).append("'");
- }
- String s = bldr.toString();
+ }
+ String s = bldr.toString();
/*String s = "SELECT COLUMN_NAME FROM " + (ci.partName == null ? "TAB_COL_STATS" :
"PART_COL_STATS")
+ " WHERE DB_NAME='" + ci.dbname + "' AND TABLE_NAME='" + ci.tableName + "'"
+ (ci.partName == null ? "" : " AND PARTITION_NAME='" + ci.partName + "'");*/
- LOG.debug("Going to execute <" + s + ">");
- rs = stmt.executeQuery(s);
- List<String> columns = new ArrayList<String>();
- while(rs.next()) {
- columns.add(rs.getString(1));
- }
- LOG.debug("Found columns to update stats: " + columns + " on " + ci.tableName +
- (ci.partName == null ? "" : "/" + ci.partName));
- dbConn.commit();
- return columns;
- } catch (SQLException e) {
- try {
+ LOG.debug("Going to execute <" + s + ">");
+ rs = stmt.executeQuery(s);
+ List<String> columns = new ArrayList<String>();
+ while (rs.next()) {
+ columns.add(rs.getString(1));
+ }
+ LOG.debug("Found columns to update stats: " + columns + " on " + ci.tableName +
+ (ci.partName == null ? "" : "/" + ci.partName));
+ dbConn.commit();
+ return columns;
+ } catch (SQLException e) {
LOG.error("Failed to find columns to analyze stats on for " + ci.tableName +
- (ci.partName == null ? "" : "/" + ci.partName), e);
- dbConn.rollback();
- } catch (SQLException e1) {
- //nothing we can do here
+ (ci.partName == null ? "" : "/" + ci.partName), e);
+ rollbackDBConn(dbConn);
+ checkRetryable(dbConn, e, "findColumnsWithStats");
+ throw new MetaException("Unable to connect to transaction database " +
+ StringUtils.stringifyException(e));
+ } finally {
+ close(rs, stmt, dbConn);
}
- throw new MetaException("Unable to connect to transaction database " +
- StringUtils.stringifyException(e));
- } finally {
- close(rs, stmt, dbConn);
+ } catch (RetryException ex) {
+ return findColumnsWithStats(ci);
}
}
}