You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by sa...@apache.org on 2018/04/17 07:24:40 UTC
[02/12] hive git commit: HIVE-19089: Create/Replicate Allocate
write-id event (Mahesh Kumar Behera, reviewed by Sankar Hariappan)
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/gen/thrift/gen-py/hive_metastore/ttypes.py
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/gen/thrift/gen-py/hive_metastore/ttypes.py b/standalone-metastore/src/gen/thrift/gen-py/hive_metastore/ttypes.py
index f2f61e0..9bf9843 100644
--- a/standalone-metastore/src/gen/thrift/gen-py/hive_metastore/ttypes.py
+++ b/standalone-metastore/src/gen/thrift/gen-py/hive_metastore/ttypes.py
@@ -11834,22 +11834,28 @@ class GetValidWriteIdsResponse:
class AllocateTableWriteIdsRequest:
"""
Attributes:
- - txnIds
- dbName
- tableName
+ - txnIds
+ - replPolicy
+ - srcTxnToWriteIdList
"""
thrift_spec = (
None, # 0
- (1, TType.LIST, 'txnIds', (TType.I64,None), None, ), # 1
- (2, TType.STRING, 'dbName', None, None, ), # 2
- (3, TType.STRING, 'tableName', None, None, ), # 3
+ (1, TType.STRING, 'dbName', None, None, ), # 1
+ (2, TType.STRING, 'tableName', None, None, ), # 2
+ (3, TType.LIST, 'txnIds', (TType.I64,None), None, ), # 3
+ (4, TType.STRING, 'replPolicy', None, None, ), # 4
+ (5, TType.LIST, 'srcTxnToWriteIdList', (TType.STRUCT,(TxnToWriteId, TxnToWriteId.thrift_spec)), None, ), # 5
)
- def __init__(self, txnIds=None, dbName=None, tableName=None,):
- self.txnIds = txnIds
+ def __init__(self, dbName=None, tableName=None, txnIds=None, replPolicy=None, srcTxnToWriteIdList=None,):
self.dbName = dbName
self.tableName = tableName
+ self.txnIds = txnIds
+ self.replPolicy = replPolicy
+ self.srcTxnToWriteIdList = srcTxnToWriteIdList
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:
@@ -11861,6 +11867,16 @@ class AllocateTableWriteIdsRequest:
if ftype == TType.STOP:
break
if fid == 1:
+ if ftype == TType.STRING:
+ self.dbName = iprot.readString()
+ else:
+ iprot.skip(ftype)
+ elif fid == 2:
+ if ftype == TType.STRING:
+ self.tableName = iprot.readString()
+ else:
+ iprot.skip(ftype)
+ elif fid == 3:
if ftype == TType.LIST:
self.txnIds = []
(_etype547, _size544) = iprot.readListBegin()
@@ -11870,14 +11886,20 @@ class AllocateTableWriteIdsRequest:
iprot.readListEnd()
else:
iprot.skip(ftype)
- elif fid == 2:
+ elif fid == 4:
if ftype == TType.STRING:
- self.dbName = iprot.readString()
+ self.replPolicy = iprot.readString()
else:
iprot.skip(ftype)
- elif fid == 3:
- if ftype == TType.STRING:
- self.tableName = iprot.readString()
+ elif fid == 5:
+ if ftype == TType.LIST:
+ self.srcTxnToWriteIdList = []
+ (_etype553, _size550) = iprot.readListBegin()
+ for _i554 in xrange(_size550):
+ _elem555 = TxnToWriteId()
+ _elem555.read(iprot)
+ self.srcTxnToWriteIdList.append(_elem555)
+ iprot.readListEnd()
else:
iprot.skip(ftype)
else:
@@ -11890,27 +11912,36 @@ class AllocateTableWriteIdsRequest:
oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec)))
return
oprot.writeStructBegin('AllocateTableWriteIdsRequest')
- if self.txnIds is not None:
- oprot.writeFieldBegin('txnIds', TType.LIST, 1)
- oprot.writeListBegin(TType.I64, len(self.txnIds))
- for iter550 in self.txnIds:
- oprot.writeI64(iter550)
- oprot.writeListEnd()
- oprot.writeFieldEnd()
if self.dbName is not None:
- oprot.writeFieldBegin('dbName', TType.STRING, 2)
+ oprot.writeFieldBegin('dbName', TType.STRING, 1)
oprot.writeString(self.dbName)
oprot.writeFieldEnd()
if self.tableName is not None:
- oprot.writeFieldBegin('tableName', TType.STRING, 3)
+ oprot.writeFieldBegin('tableName', TType.STRING, 2)
oprot.writeString(self.tableName)
oprot.writeFieldEnd()
+ if self.txnIds is not None:
+ oprot.writeFieldBegin('txnIds', TType.LIST, 3)
+ oprot.writeListBegin(TType.I64, len(self.txnIds))
+ for iter556 in self.txnIds:
+ oprot.writeI64(iter556)
+ oprot.writeListEnd()
+ oprot.writeFieldEnd()
+ if self.replPolicy is not None:
+ oprot.writeFieldBegin('replPolicy', TType.STRING, 4)
+ oprot.writeString(self.replPolicy)
+ oprot.writeFieldEnd()
+ if self.srcTxnToWriteIdList is not None:
+ oprot.writeFieldBegin('srcTxnToWriteIdList', TType.LIST, 5)
+ oprot.writeListBegin(TType.STRUCT, len(self.srcTxnToWriteIdList))
+ for iter557 in self.srcTxnToWriteIdList:
+ iter557.write(oprot)
+ oprot.writeListEnd()
+ oprot.writeFieldEnd()
oprot.writeFieldStop()
oprot.writeStructEnd()
def validate(self):
- if self.txnIds is None:
- raise TProtocol.TProtocolException(message='Required field txnIds is unset!')
if self.dbName is None:
raise TProtocol.TProtocolException(message='Required field dbName is unset!')
if self.tableName is None:
@@ -11920,9 +11951,11 @@ class AllocateTableWriteIdsRequest:
def __hash__(self):
value = 17
- value = (value * 31) ^ hash(self.txnIds)
value = (value * 31) ^ hash(self.dbName)
value = (value * 31) ^ hash(self.tableName)
+ value = (value * 31) ^ hash(self.txnIds)
+ value = (value * 31) ^ hash(self.replPolicy)
+ value = (value * 31) ^ hash(self.srcTxnToWriteIdList)
return value
def __repr__(self):
@@ -12044,11 +12077,11 @@ class AllocateTableWriteIdsResponse:
if fid == 1:
if ftype == TType.LIST:
self.txnToWriteIds = []
- (_etype554, _size551) = iprot.readListBegin()
- for _i555 in xrange(_size551):
- _elem556 = TxnToWriteId()
- _elem556.read(iprot)
- self.txnToWriteIds.append(_elem556)
+ (_etype561, _size558) = iprot.readListBegin()
+ for _i562 in xrange(_size558):
+ _elem563 = TxnToWriteId()
+ _elem563.read(iprot)
+ self.txnToWriteIds.append(_elem563)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -12065,8 +12098,8 @@ class AllocateTableWriteIdsResponse:
if self.txnToWriteIds is not None:
oprot.writeFieldBegin('txnToWriteIds', TType.LIST, 1)
oprot.writeListBegin(TType.STRUCT, len(self.txnToWriteIds))
- for iter557 in self.txnToWriteIds:
- iter557.write(oprot)
+ for iter564 in self.txnToWriteIds:
+ iter564.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -12294,11 +12327,11 @@ class LockRequest:
if fid == 1:
if ftype == TType.LIST:
self.component = []
- (_etype561, _size558) = iprot.readListBegin()
- for _i562 in xrange(_size558):
- _elem563 = LockComponent()
- _elem563.read(iprot)
- self.component.append(_elem563)
+ (_etype568, _size565) = iprot.readListBegin()
+ for _i569 in xrange(_size565):
+ _elem570 = LockComponent()
+ _elem570.read(iprot)
+ self.component.append(_elem570)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -12335,8 +12368,8 @@ class LockRequest:
if self.component is not None:
oprot.writeFieldBegin('component', TType.LIST, 1)
oprot.writeListBegin(TType.STRUCT, len(self.component))
- for iter564 in self.component:
- iter564.write(oprot)
+ for iter571 in self.component:
+ iter571.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.txnid is not None:
@@ -13034,11 +13067,11 @@ class ShowLocksResponse:
if fid == 1:
if ftype == TType.LIST:
self.locks = []
- (_etype568, _size565) = iprot.readListBegin()
- for _i569 in xrange(_size565):
- _elem570 = ShowLocksResponseElement()
- _elem570.read(iprot)
- self.locks.append(_elem570)
+ (_etype575, _size572) = iprot.readListBegin()
+ for _i576 in xrange(_size572):
+ _elem577 = ShowLocksResponseElement()
+ _elem577.read(iprot)
+ self.locks.append(_elem577)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -13055,8 +13088,8 @@ class ShowLocksResponse:
if self.locks is not None:
oprot.writeFieldBegin('locks', TType.LIST, 1)
oprot.writeListBegin(TType.STRUCT, len(self.locks))
- for iter571 in self.locks:
- iter571.write(oprot)
+ for iter578 in self.locks:
+ iter578.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -13271,20 +13304,20 @@ class HeartbeatTxnRangeResponse:
if fid == 1:
if ftype == TType.SET:
self.aborted = set()
- (_etype575, _size572) = iprot.readSetBegin()
- for _i576 in xrange(_size572):
- _elem577 = iprot.readI64()
- self.aborted.add(_elem577)
+ (_etype582, _size579) = iprot.readSetBegin()
+ for _i583 in xrange(_size579):
+ _elem584 = iprot.readI64()
+ self.aborted.add(_elem584)
iprot.readSetEnd()
else:
iprot.skip(ftype)
elif fid == 2:
if ftype == TType.SET:
self.nosuch = set()
- (_etype581, _size578) = iprot.readSetBegin()
- for _i582 in xrange(_size578):
- _elem583 = iprot.readI64()
- self.nosuch.add(_elem583)
+ (_etype588, _size585) = iprot.readSetBegin()
+ for _i589 in xrange(_size585):
+ _elem590 = iprot.readI64()
+ self.nosuch.add(_elem590)
iprot.readSetEnd()
else:
iprot.skip(ftype)
@@ -13301,15 +13334,15 @@ class HeartbeatTxnRangeResponse:
if self.aborted is not None:
oprot.writeFieldBegin('aborted', TType.SET, 1)
oprot.writeSetBegin(TType.I64, len(self.aborted))
- for iter584 in self.aborted:
- oprot.writeI64(iter584)
+ for iter591 in self.aborted:
+ oprot.writeI64(iter591)
oprot.writeSetEnd()
oprot.writeFieldEnd()
if self.nosuch is not None:
oprot.writeFieldBegin('nosuch', TType.SET, 2)
oprot.writeSetBegin(TType.I64, len(self.nosuch))
- for iter585 in self.nosuch:
- oprot.writeI64(iter585)
+ for iter592 in self.nosuch:
+ oprot.writeI64(iter592)
oprot.writeSetEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -13406,11 +13439,11 @@ class CompactionRequest:
elif fid == 6:
if ftype == TType.MAP:
self.properties = {}
- (_ktype587, _vtype588, _size586 ) = iprot.readMapBegin()
- for _i590 in xrange(_size586):
- _key591 = iprot.readString()
- _val592 = iprot.readString()
- self.properties[_key591] = _val592
+ (_ktype594, _vtype595, _size593 ) = iprot.readMapBegin()
+ for _i597 in xrange(_size593):
+ _key598 = iprot.readString()
+ _val599 = iprot.readString()
+ self.properties[_key598] = _val599
iprot.readMapEnd()
else:
iprot.skip(ftype)
@@ -13447,9 +13480,9 @@ class CompactionRequest:
if self.properties is not None:
oprot.writeFieldBegin('properties', TType.MAP, 6)
oprot.writeMapBegin(TType.STRING, TType.STRING, len(self.properties))
- for kiter593,viter594 in self.properties.items():
- oprot.writeString(kiter593)
- oprot.writeString(viter594)
+ for kiter600,viter601 in self.properties.items():
+ oprot.writeString(kiter600)
+ oprot.writeString(viter601)
oprot.writeMapEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -13884,11 +13917,11 @@ class ShowCompactResponse:
if fid == 1:
if ftype == TType.LIST:
self.compacts = []
- (_etype598, _size595) = iprot.readListBegin()
- for _i599 in xrange(_size595):
- _elem600 = ShowCompactResponseElement()
- _elem600.read(iprot)
- self.compacts.append(_elem600)
+ (_etype605, _size602) = iprot.readListBegin()
+ for _i606 in xrange(_size602):
+ _elem607 = ShowCompactResponseElement()
+ _elem607.read(iprot)
+ self.compacts.append(_elem607)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -13905,8 +13938,8 @@ class ShowCompactResponse:
if self.compacts is not None:
oprot.writeFieldBegin('compacts', TType.LIST, 1)
oprot.writeListBegin(TType.STRUCT, len(self.compacts))
- for iter601 in self.compacts:
- iter601.write(oprot)
+ for iter608 in self.compacts:
+ iter608.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -13995,10 +14028,10 @@ class AddDynamicPartitions:
elif fid == 5:
if ftype == TType.LIST:
self.partitionnames = []
- (_etype605, _size602) = iprot.readListBegin()
- for _i606 in xrange(_size602):
- _elem607 = iprot.readString()
- self.partitionnames.append(_elem607)
+ (_etype612, _size609) = iprot.readListBegin()
+ for _i613 in xrange(_size609):
+ _elem614 = iprot.readString()
+ self.partitionnames.append(_elem614)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -14036,8 +14069,8 @@ class AddDynamicPartitions:
if self.partitionnames is not None:
oprot.writeFieldBegin('partitionnames', TType.LIST, 5)
oprot.writeListBegin(TType.STRING, len(self.partitionnames))
- for iter608 in self.partitionnames:
- oprot.writeString(iter608)
+ for iter615 in self.partitionnames:
+ oprot.writeString(iter615)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.operationType is not None:
@@ -14267,10 +14300,10 @@ class CreationMetadata:
elif fid == 4:
if ftype == TType.SET:
self.tablesUsed = set()
- (_etype612, _size609) = iprot.readSetBegin()
- for _i613 in xrange(_size609):
- _elem614 = iprot.readString()
- self.tablesUsed.add(_elem614)
+ (_etype619, _size616) = iprot.readSetBegin()
+ for _i620 in xrange(_size616):
+ _elem621 = iprot.readString()
+ self.tablesUsed.add(_elem621)
iprot.readSetEnd()
else:
iprot.skip(ftype)
@@ -14304,8 +14337,8 @@ class CreationMetadata:
if self.tablesUsed is not None:
oprot.writeFieldBegin('tablesUsed', TType.SET, 4)
oprot.writeSetBegin(TType.STRING, len(self.tablesUsed))
- for iter615 in self.tablesUsed:
- oprot.writeString(iter615)
+ for iter622 in self.tablesUsed:
+ oprot.writeString(iter622)
oprot.writeSetEnd()
oprot.writeFieldEnd()
if self.validTxnList is not None:
@@ -14617,11 +14650,11 @@ class NotificationEventResponse:
if fid == 1:
if ftype == TType.LIST:
self.events = []
- (_etype619, _size616) = iprot.readListBegin()
- for _i620 in xrange(_size616):
- _elem621 = NotificationEvent()
- _elem621.read(iprot)
- self.events.append(_elem621)
+ (_etype626, _size623) = iprot.readListBegin()
+ for _i627 in xrange(_size623):
+ _elem628 = NotificationEvent()
+ _elem628.read(iprot)
+ self.events.append(_elem628)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -14638,8 +14671,8 @@ class NotificationEventResponse:
if self.events is not None:
oprot.writeFieldBegin('events', TType.LIST, 1)
oprot.writeListBegin(TType.STRUCT, len(self.events))
- for iter622 in self.events:
- iter622.write(oprot)
+ for iter629 in self.events:
+ iter629.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -14933,20 +14966,20 @@ class InsertEventRequestData:
elif fid == 2:
if ftype == TType.LIST:
self.filesAdded = []
- (_etype626, _size623) = iprot.readListBegin()
- for _i627 in xrange(_size623):
- _elem628 = iprot.readString()
- self.filesAdded.append(_elem628)
+ (_etype633, _size630) = iprot.readListBegin()
+ for _i634 in xrange(_size630):
+ _elem635 = iprot.readString()
+ self.filesAdded.append(_elem635)
iprot.readListEnd()
else:
iprot.skip(ftype)
elif fid == 3:
if ftype == TType.LIST:
self.filesAddedChecksum = []
- (_etype632, _size629) = iprot.readListBegin()
- for _i633 in xrange(_size629):
- _elem634 = iprot.readString()
- self.filesAddedChecksum.append(_elem634)
+ (_etype639, _size636) = iprot.readListBegin()
+ for _i640 in xrange(_size636):
+ _elem641 = iprot.readString()
+ self.filesAddedChecksum.append(_elem641)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -14967,15 +15000,15 @@ class InsertEventRequestData:
if self.filesAdded is not None:
oprot.writeFieldBegin('filesAdded', TType.LIST, 2)
oprot.writeListBegin(TType.STRING, len(self.filesAdded))
- for iter635 in self.filesAdded:
- oprot.writeString(iter635)
+ for iter642 in self.filesAdded:
+ oprot.writeString(iter642)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.filesAddedChecksum is not None:
oprot.writeFieldBegin('filesAddedChecksum', TType.LIST, 3)
oprot.writeListBegin(TType.STRING, len(self.filesAddedChecksum))
- for iter636 in self.filesAddedChecksum:
- oprot.writeString(iter636)
+ for iter643 in self.filesAddedChecksum:
+ oprot.writeString(iter643)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -15133,10 +15166,10 @@ class FireEventRequest:
elif fid == 5:
if ftype == TType.LIST:
self.partitionVals = []
- (_etype640, _size637) = iprot.readListBegin()
- for _i641 in xrange(_size637):
- _elem642 = iprot.readString()
- self.partitionVals.append(_elem642)
+ (_etype647, _size644) = iprot.readListBegin()
+ for _i648 in xrange(_size644):
+ _elem649 = iprot.readString()
+ self.partitionVals.append(_elem649)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -15174,8 +15207,8 @@ class FireEventRequest:
if self.partitionVals is not None:
oprot.writeFieldBegin('partitionVals', TType.LIST, 5)
oprot.writeListBegin(TType.STRING, len(self.partitionVals))
- for iter643 in self.partitionVals:
- oprot.writeString(iter643)
+ for iter650 in self.partitionVals:
+ oprot.writeString(iter650)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.catName is not None:
@@ -15367,12 +15400,12 @@ class GetFileMetadataByExprResult:
if fid == 1:
if ftype == TType.MAP:
self.metadata = {}
- (_ktype645, _vtype646, _size644 ) = iprot.readMapBegin()
- for _i648 in xrange(_size644):
- _key649 = iprot.readI64()
- _val650 = MetadataPpdResult()
- _val650.read(iprot)
- self.metadata[_key649] = _val650
+ (_ktype652, _vtype653, _size651 ) = iprot.readMapBegin()
+ for _i655 in xrange(_size651):
+ _key656 = iprot.readI64()
+ _val657 = MetadataPpdResult()
+ _val657.read(iprot)
+ self.metadata[_key656] = _val657
iprot.readMapEnd()
else:
iprot.skip(ftype)
@@ -15394,9 +15427,9 @@ class GetFileMetadataByExprResult:
if self.metadata is not None:
oprot.writeFieldBegin('metadata', TType.MAP, 1)
oprot.writeMapBegin(TType.I64, TType.STRUCT, len(self.metadata))
- for kiter651,viter652 in self.metadata.items():
- oprot.writeI64(kiter651)
- viter652.write(oprot)
+ for kiter658,viter659 in self.metadata.items():
+ oprot.writeI64(kiter658)
+ viter659.write(oprot)
oprot.writeMapEnd()
oprot.writeFieldEnd()
if self.isSupported is not None:
@@ -15466,10 +15499,10 @@ class GetFileMetadataByExprRequest:
if fid == 1:
if ftype == TType.LIST:
self.fileIds = []
- (_etype656, _size653) = iprot.readListBegin()
- for _i657 in xrange(_size653):
- _elem658 = iprot.readI64()
- self.fileIds.append(_elem658)
+ (_etype663, _size660) = iprot.readListBegin()
+ for _i664 in xrange(_size660):
+ _elem665 = iprot.readI64()
+ self.fileIds.append(_elem665)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -15501,8 +15534,8 @@ class GetFileMetadataByExprRequest:
if self.fileIds is not None:
oprot.writeFieldBegin('fileIds', TType.LIST, 1)
oprot.writeListBegin(TType.I64, len(self.fileIds))
- for iter659 in self.fileIds:
- oprot.writeI64(iter659)
+ for iter666 in self.fileIds:
+ oprot.writeI64(iter666)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.expr is not None:
@@ -15576,11 +15609,11 @@ class GetFileMetadataResult:
if fid == 1:
if ftype == TType.MAP:
self.metadata = {}
- (_ktype661, _vtype662, _size660 ) = iprot.readMapBegin()
- for _i664 in xrange(_size660):
- _key665 = iprot.readI64()
- _val666 = iprot.readString()
- self.metadata[_key665] = _val666
+ (_ktype668, _vtype669, _size667 ) = iprot.readMapBegin()
+ for _i671 in xrange(_size667):
+ _key672 = iprot.readI64()
+ _val673 = iprot.readString()
+ self.metadata[_key672] = _val673
iprot.readMapEnd()
else:
iprot.skip(ftype)
@@ -15602,9 +15635,9 @@ class GetFileMetadataResult:
if self.metadata is not None:
oprot.writeFieldBegin('metadata', TType.MAP, 1)
oprot.writeMapBegin(TType.I64, TType.STRING, len(self.metadata))
- for kiter667,viter668 in self.metadata.items():
- oprot.writeI64(kiter667)
- oprot.writeString(viter668)
+ for kiter674,viter675 in self.metadata.items():
+ oprot.writeI64(kiter674)
+ oprot.writeString(viter675)
oprot.writeMapEnd()
oprot.writeFieldEnd()
if self.isSupported is not None:
@@ -15665,10 +15698,10 @@ class GetFileMetadataRequest:
if fid == 1:
if ftype == TType.LIST:
self.fileIds = []
- (_etype672, _size669) = iprot.readListBegin()
- for _i673 in xrange(_size669):
- _elem674 = iprot.readI64()
- self.fileIds.append(_elem674)
+ (_etype679, _size676) = iprot.readListBegin()
+ for _i680 in xrange(_size676):
+ _elem681 = iprot.readI64()
+ self.fileIds.append(_elem681)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -15685,8 +15718,8 @@ class GetFileMetadataRequest:
if self.fileIds is not None:
oprot.writeFieldBegin('fileIds', TType.LIST, 1)
oprot.writeListBegin(TType.I64, len(self.fileIds))
- for iter675 in self.fileIds:
- oprot.writeI64(iter675)
+ for iter682 in self.fileIds:
+ oprot.writeI64(iter682)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -15792,20 +15825,20 @@ class PutFileMetadataRequest:
if fid == 1:
if ftype == TType.LIST:
self.fileIds = []
- (_etype679, _size676) = iprot.readListBegin()
- for _i680 in xrange(_size676):
- _elem681 = iprot.readI64()
- self.fileIds.append(_elem681)
+ (_etype686, _size683) = iprot.readListBegin()
+ for _i687 in xrange(_size683):
+ _elem688 = iprot.readI64()
+ self.fileIds.append(_elem688)
iprot.readListEnd()
else:
iprot.skip(ftype)
elif fid == 2:
if ftype == TType.LIST:
self.metadata = []
- (_etype685, _size682) = iprot.readListBegin()
- for _i686 in xrange(_size682):
- _elem687 = iprot.readString()
- self.metadata.append(_elem687)
+ (_etype692, _size689) = iprot.readListBegin()
+ for _i693 in xrange(_size689):
+ _elem694 = iprot.readString()
+ self.metadata.append(_elem694)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -15827,15 +15860,15 @@ class PutFileMetadataRequest:
if self.fileIds is not None:
oprot.writeFieldBegin('fileIds', TType.LIST, 1)
oprot.writeListBegin(TType.I64, len(self.fileIds))
- for iter688 in self.fileIds:
- oprot.writeI64(iter688)
+ for iter695 in self.fileIds:
+ oprot.writeI64(iter695)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.metadata is not None:
oprot.writeFieldBegin('metadata', TType.LIST, 2)
oprot.writeListBegin(TType.STRING, len(self.metadata))
- for iter689 in self.metadata:
- oprot.writeString(iter689)
+ for iter696 in self.metadata:
+ oprot.writeString(iter696)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.type is not None:
@@ -15943,10 +15976,10 @@ class ClearFileMetadataRequest:
if fid == 1:
if ftype == TType.LIST:
self.fileIds = []
- (_etype693, _size690) = iprot.readListBegin()
- for _i694 in xrange(_size690):
- _elem695 = iprot.readI64()
- self.fileIds.append(_elem695)
+ (_etype700, _size697) = iprot.readListBegin()
+ for _i701 in xrange(_size697):
+ _elem702 = iprot.readI64()
+ self.fileIds.append(_elem702)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -15963,8 +15996,8 @@ class ClearFileMetadataRequest:
if self.fileIds is not None:
oprot.writeFieldBegin('fileIds', TType.LIST, 1)
oprot.writeListBegin(TType.I64, len(self.fileIds))
- for iter696 in self.fileIds:
- oprot.writeI64(iter696)
+ for iter703 in self.fileIds:
+ oprot.writeI64(iter703)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -16193,11 +16226,11 @@ class GetAllFunctionsResponse:
if fid == 1:
if ftype == TType.LIST:
self.functions = []
- (_etype700, _size697) = iprot.readListBegin()
- for _i701 in xrange(_size697):
- _elem702 = Function()
- _elem702.read(iprot)
- self.functions.append(_elem702)
+ (_etype707, _size704) = iprot.readListBegin()
+ for _i708 in xrange(_size704):
+ _elem709 = Function()
+ _elem709.read(iprot)
+ self.functions.append(_elem709)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -16214,8 +16247,8 @@ class GetAllFunctionsResponse:
if self.functions is not None:
oprot.writeFieldBegin('functions', TType.LIST, 1)
oprot.writeListBegin(TType.STRUCT, len(self.functions))
- for iter703 in self.functions:
- iter703.write(oprot)
+ for iter710 in self.functions:
+ iter710.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -16267,10 +16300,10 @@ class ClientCapabilities:
if fid == 1:
if ftype == TType.LIST:
self.values = []
- (_etype707, _size704) = iprot.readListBegin()
- for _i708 in xrange(_size704):
- _elem709 = iprot.readI32()
- self.values.append(_elem709)
+ (_etype714, _size711) = iprot.readListBegin()
+ for _i715 in xrange(_size711):
+ _elem716 = iprot.readI32()
+ self.values.append(_elem716)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -16287,8 +16320,8 @@ class ClientCapabilities:
if self.values is not None:
oprot.writeFieldBegin('values', TType.LIST, 1)
oprot.writeListBegin(TType.I32, len(self.values))
- for iter710 in self.values:
- oprot.writeI32(iter710)
+ for iter717 in self.values:
+ oprot.writeI32(iter717)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -16533,10 +16566,10 @@ class GetTablesRequest:
elif fid == 2:
if ftype == TType.LIST:
self.tblNames = []
- (_etype714, _size711) = iprot.readListBegin()
- for _i715 in xrange(_size711):
- _elem716 = iprot.readString()
- self.tblNames.append(_elem716)
+ (_etype721, _size718) = iprot.readListBegin()
+ for _i722 in xrange(_size718):
+ _elem723 = iprot.readString()
+ self.tblNames.append(_elem723)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -16568,8 +16601,8 @@ class GetTablesRequest:
if self.tblNames is not None:
oprot.writeFieldBegin('tblNames', TType.LIST, 2)
oprot.writeListBegin(TType.STRING, len(self.tblNames))
- for iter717 in self.tblNames:
- oprot.writeString(iter717)
+ for iter724 in self.tblNames:
+ oprot.writeString(iter724)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.capabilities is not None:
@@ -16634,11 +16667,11 @@ class GetTablesResult:
if fid == 1:
if ftype == TType.LIST:
self.tables = []
- (_etype721, _size718) = iprot.readListBegin()
- for _i722 in xrange(_size718):
- _elem723 = Table()
- _elem723.read(iprot)
- self.tables.append(_elem723)
+ (_etype728, _size725) = iprot.readListBegin()
+ for _i729 in xrange(_size725):
+ _elem730 = Table()
+ _elem730.read(iprot)
+ self.tables.append(_elem730)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -16655,8 +16688,8 @@ class GetTablesResult:
if self.tables is not None:
oprot.writeFieldBegin('tables', TType.LIST, 1)
oprot.writeListBegin(TType.STRUCT, len(self.tables))
- for iter724 in self.tables:
- iter724.write(oprot)
+ for iter731 in self.tables:
+ iter731.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -16970,10 +17003,10 @@ class Materialization:
if fid == 1:
if ftype == TType.SET:
self.tablesUsed = set()
- (_etype728, _size725) = iprot.readSetBegin()
- for _i729 in xrange(_size725):
- _elem730 = iprot.readString()
- self.tablesUsed.add(_elem730)
+ (_etype735, _size732) = iprot.readSetBegin()
+ for _i736 in xrange(_size732):
+ _elem737 = iprot.readString()
+ self.tablesUsed.add(_elem737)
iprot.readSetEnd()
else:
iprot.skip(ftype)
@@ -17005,8 +17038,8 @@ class Materialization:
if self.tablesUsed is not None:
oprot.writeFieldBegin('tablesUsed', TType.SET, 1)
oprot.writeSetBegin(TType.STRING, len(self.tablesUsed))
- for iter731 in self.tablesUsed:
- oprot.writeString(iter731)
+ for iter738 in self.tablesUsed:
+ oprot.writeString(iter738)
oprot.writeSetEnd()
oprot.writeFieldEnd()
if self.validTxnList is not None:
@@ -17911,44 +17944,44 @@ class WMFullResourcePlan:
elif fid == 2:
if ftype == TType.LIST:
self.pools = []
- (_etype735, _size732) = iprot.readListBegin()
- for _i736 in xrange(_size732):
- _elem737 = WMPool()
- _elem737.read(iprot)
- self.pools.append(_elem737)
+ (_etype742, _size739) = iprot.readListBegin()
+ for _i743 in xrange(_size739):
+ _elem744 = WMPool()
+ _elem744.read(iprot)
+ self.pools.append(_elem744)
iprot.readListEnd()
else:
iprot.skip(ftype)
elif fid == 3:
if ftype == TType.LIST:
self.mappings = []
- (_etype741, _size738) = iprot.readListBegin()
- for _i742 in xrange(_size738):
- _elem743 = WMMapping()
- _elem743.read(iprot)
- self.mappings.append(_elem743)
+ (_etype748, _size745) = iprot.readListBegin()
+ for _i749 in xrange(_size745):
+ _elem750 = WMMapping()
+ _elem750.read(iprot)
+ self.mappings.append(_elem750)
iprot.readListEnd()
else:
iprot.skip(ftype)
elif fid == 4:
if ftype == TType.LIST:
self.triggers = []
- (_etype747, _size744) = iprot.readListBegin()
- for _i748 in xrange(_size744):
- _elem749 = WMTrigger()
- _elem749.read(iprot)
- self.triggers.append(_elem749)
+ (_etype754, _size751) = iprot.readListBegin()
+ for _i755 in xrange(_size751):
+ _elem756 = WMTrigger()
+ _elem756.read(iprot)
+ self.triggers.append(_elem756)
iprot.readListEnd()
else:
iprot.skip(ftype)
elif fid == 5:
if ftype == TType.LIST:
self.poolTriggers = []
- (_etype753, _size750) = iprot.readListBegin()
- for _i754 in xrange(_size750):
- _elem755 = WMPoolTrigger()
- _elem755.read(iprot)
- self.poolTriggers.append(_elem755)
+ (_etype760, _size757) = iprot.readListBegin()
+ for _i761 in xrange(_size757):
+ _elem762 = WMPoolTrigger()
+ _elem762.read(iprot)
+ self.poolTriggers.append(_elem762)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -17969,29 +18002,29 @@ class WMFullResourcePlan:
if self.pools is not None:
oprot.writeFieldBegin('pools', TType.LIST, 2)
oprot.writeListBegin(TType.STRUCT, len(self.pools))
- for iter756 in self.pools:
- iter756.write(oprot)
+ for iter763 in self.pools:
+ iter763.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.mappings is not None:
oprot.writeFieldBegin('mappings', TType.LIST, 3)
oprot.writeListBegin(TType.STRUCT, len(self.mappings))
- for iter757 in self.mappings:
- iter757.write(oprot)
+ for iter764 in self.mappings:
+ iter764.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.triggers is not None:
oprot.writeFieldBegin('triggers', TType.LIST, 4)
oprot.writeListBegin(TType.STRUCT, len(self.triggers))
- for iter758 in self.triggers:
- iter758.write(oprot)
+ for iter765 in self.triggers:
+ iter765.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.poolTriggers is not None:
oprot.writeFieldBegin('poolTriggers', TType.LIST, 5)
oprot.writeListBegin(TType.STRUCT, len(self.poolTriggers))
- for iter759 in self.poolTriggers:
- iter759.write(oprot)
+ for iter766 in self.poolTriggers:
+ iter766.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -18465,11 +18498,11 @@ class WMGetAllResourcePlanResponse:
if fid == 1:
if ftype == TType.LIST:
self.resourcePlans = []
- (_etype763, _size760) = iprot.readListBegin()
- for _i764 in xrange(_size760):
- _elem765 = WMResourcePlan()
- _elem765.read(iprot)
- self.resourcePlans.append(_elem765)
+ (_etype770, _size767) = iprot.readListBegin()
+ for _i771 in xrange(_size767):
+ _elem772 = WMResourcePlan()
+ _elem772.read(iprot)
+ self.resourcePlans.append(_elem772)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -18486,8 +18519,8 @@ class WMGetAllResourcePlanResponse:
if self.resourcePlans is not None:
oprot.writeFieldBegin('resourcePlans', TType.LIST, 1)
oprot.writeListBegin(TType.STRUCT, len(self.resourcePlans))
- for iter766 in self.resourcePlans:
- iter766.write(oprot)
+ for iter773 in self.resourcePlans:
+ iter773.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -18791,20 +18824,20 @@ class WMValidateResourcePlanResponse:
if fid == 1:
if ftype == TType.LIST:
self.errors = []
- (_etype770, _size767) = iprot.readListBegin()
- for _i771 in xrange(_size767):
- _elem772 = iprot.readString()
- self.errors.append(_elem772)
+ (_etype777, _size774) = iprot.readListBegin()
+ for _i778 in xrange(_size774):
+ _elem779 = iprot.readString()
+ self.errors.append(_elem779)
iprot.readListEnd()
else:
iprot.skip(ftype)
elif fid == 2:
if ftype == TType.LIST:
self.warnings = []
- (_etype776, _size773) = iprot.readListBegin()
- for _i777 in xrange(_size773):
- _elem778 = iprot.readString()
- self.warnings.append(_elem778)
+ (_etype783, _size780) = iprot.readListBegin()
+ for _i784 in xrange(_size780):
+ _elem785 = iprot.readString()
+ self.warnings.append(_elem785)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -18821,15 +18854,15 @@ class WMValidateResourcePlanResponse:
if self.errors is not None:
oprot.writeFieldBegin('errors', TType.LIST, 1)
oprot.writeListBegin(TType.STRING, len(self.errors))
- for iter779 in self.errors:
- oprot.writeString(iter779)
+ for iter786 in self.errors:
+ oprot.writeString(iter786)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.warnings is not None:
oprot.writeFieldBegin('warnings', TType.LIST, 2)
oprot.writeListBegin(TType.STRING, len(self.warnings))
- for iter780 in self.warnings:
- oprot.writeString(iter780)
+ for iter787 in self.warnings:
+ oprot.writeString(iter787)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -19406,11 +19439,11 @@ class WMGetTriggersForResourePlanResponse:
if fid == 1:
if ftype == TType.LIST:
self.triggers = []
- (_etype784, _size781) = iprot.readListBegin()
- for _i785 in xrange(_size781):
- _elem786 = WMTrigger()
- _elem786.read(iprot)
- self.triggers.append(_elem786)
+ (_etype791, _size788) = iprot.readListBegin()
+ for _i792 in xrange(_size788):
+ _elem793 = WMTrigger()
+ _elem793.read(iprot)
+ self.triggers.append(_elem793)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -19427,8 +19460,8 @@ class WMGetTriggersForResourePlanResponse:
if self.triggers is not None:
oprot.writeFieldBegin('triggers', TType.LIST, 1)
oprot.writeListBegin(TType.STRUCT, len(self.triggers))
- for iter787 in self.triggers:
- iter787.write(oprot)
+ for iter794 in self.triggers:
+ iter794.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
@@ -20612,11 +20645,11 @@ class SchemaVersion:
elif fid == 4:
if ftype == TType.LIST:
self.cols = []
- (_etype791, _size788) = iprot.readListBegin()
- for _i792 in xrange(_size788):
- _elem793 = FieldSchema()
- _elem793.read(iprot)
- self.cols.append(_elem793)
+ (_etype798, _size795) = iprot.readListBegin()
+ for _i799 in xrange(_size795):
+ _elem800 = FieldSchema()
+ _elem800.read(iprot)
+ self.cols.append(_elem800)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -20676,8 +20709,8 @@ class SchemaVersion:
if self.cols is not None:
oprot.writeFieldBegin('cols', TType.LIST, 4)
oprot.writeListBegin(TType.STRUCT, len(self.cols))
- for iter794 in self.cols:
- iter794.write(oprot)
+ for iter801 in self.cols:
+ iter801.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
if self.state is not None:
@@ -20932,11 +20965,11 @@ class FindSchemasByColsResp:
if fid == 1:
if ftype == TType.LIST:
self.schemaVersions = []
- (_etype798, _size795) = iprot.readListBegin()
- for _i799 in xrange(_size795):
- _elem800 = SchemaVersionDescriptor()
- _elem800.read(iprot)
- self.schemaVersions.append(_elem800)
+ (_etype805, _size802) = iprot.readListBegin()
+ for _i806 in xrange(_size802):
+ _elem807 = SchemaVersionDescriptor()
+ _elem807.read(iprot)
+ self.schemaVersions.append(_elem807)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -20953,8 +20986,8 @@ class FindSchemasByColsResp:
if self.schemaVersions is not None:
oprot.writeFieldBegin('schemaVersions', TType.LIST, 1)
oprot.writeListBegin(TType.STRUCT, len(self.schemaVersions))
- for iter801 in self.schemaVersions:
- iter801.write(oprot)
+ for iter808 in self.schemaVersions:
+ iter808.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/gen/thrift/gen-rb/hive_metastore_types.rb
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/gen/thrift/gen-rb/hive_metastore_types.rb b/standalone-metastore/src/gen/thrift/gen-rb/hive_metastore_types.rb
index 0e70e89..3dbe4d8 100644
--- a/standalone-metastore/src/gen/thrift/gen-rb/hive_metastore_types.rb
+++ b/standalone-metastore/src/gen/thrift/gen-rb/hive_metastore_types.rb
@@ -2639,20 +2639,23 @@ end
class AllocateTableWriteIdsRequest
include ::Thrift::Struct, ::Thrift::Struct_Union
- TXNIDS = 1
- DBNAME = 2
- TABLENAME = 3
+ DBNAME = 1
+ TABLENAME = 2
+ TXNIDS = 3
+ REPLPOLICY = 4
+ SRCTXNTOWRITEIDLIST = 5
FIELDS = {
- TXNIDS => {:type => ::Thrift::Types::LIST, :name => 'txnIds', :element => {:type => ::Thrift::Types::I64}},
DBNAME => {:type => ::Thrift::Types::STRING, :name => 'dbName'},
- TABLENAME => {:type => ::Thrift::Types::STRING, :name => 'tableName'}
+ TABLENAME => {:type => ::Thrift::Types::STRING, :name => 'tableName'},
+ TXNIDS => {:type => ::Thrift::Types::LIST, :name => 'txnIds', :element => {:type => ::Thrift::Types::I64}, :optional => true},
+ REPLPOLICY => {:type => ::Thrift::Types::STRING, :name => 'replPolicy', :optional => true},
+ SRCTXNTOWRITEIDLIST => {:type => ::Thrift::Types::LIST, :name => 'srcTxnToWriteIdList', :element => {:type => ::Thrift::Types::STRUCT, :class => ::TxnToWriteId}, :optional => true}
}
def struct_fields; FIELDS; end
def validate
- raise ::Thrift::ProtocolException.new(::Thrift::ProtocolException::UNKNOWN, 'Required field txnIds is unset!') unless @txnIds
raise ::Thrift::ProtocolException.new(::Thrift::ProtocolException::UNKNOWN, 'Required field dbName is unset!') unless @dbName
raise ::Thrift::ProtocolException.new(::Thrift::ProtocolException::UNKNOWN, 'Required field tableName is unset!') unless @tableName
end
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
index 565549a..ae9ec5c 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
@@ -62,7 +62,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
-import java.util.function.Consumer;
import java.util.regex.Pattern;
import javax.jdo.JDOException;
@@ -88,6 +87,7 @@ import org.apache.hadoop.hive.metastore.events.AddNotNullConstraintEvent;
import org.apache.hadoop.hive.metastore.events.AddPartitionEvent;
import org.apache.hadoop.hive.metastore.events.AddPrimaryKeyEvent;
import org.apache.hadoop.hive.metastore.events.AddUniqueConstraintEvent;
+import org.apache.hadoop.hive.metastore.events.AllocWriteIdEvent;
import org.apache.hadoop.hive.metastore.events.AlterDatabaseEvent;
import org.apache.hadoop.hive.metastore.events.AlterISchemaEvent;
import org.apache.hadoop.hive.metastore.events.AlterPartitionEvent;
@@ -177,7 +177,6 @@ import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportFactory;
-import org.iq80.leveldb.DB;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -7106,7 +7105,13 @@ public class HiveMetaStore extends ThriftHiveMetastore {
@Override
public AllocateTableWriteIdsResponse allocate_table_write_ids(
AllocateTableWriteIdsRequest rqst) throws TException {
- return getTxnHandler().allocateTableWriteIds(rqst);
+ AllocateTableWriteIdsResponse response = getTxnHandler().allocateTableWriteIds(rqst);
+ if (listeners != null && !listeners.isEmpty()) {
+ MetaStoreListenerNotifier.notifyEvent(listeners, EventType.ALLOC_WRITE_ID,
+ new AllocWriteIdEvent(response.getTxnToWriteIds(), rqst.getDbName(),
+ rqst.getTableName(), this));
+ }
+ return response;
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
index 9a43b2c..108add0 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
@@ -72,7 +72,6 @@ import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.thrift.TApplicationException;
-import org.apache.thrift.TBase;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
@@ -2436,9 +2435,22 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
@Override
public List<TxnToWriteId> allocateTableWriteIdsBatch(List<Long> txnIds, String dbName, String tableName)
throws TException {
- AllocateTableWriteIdsRequest rqst = new AllocateTableWriteIdsRequest(txnIds, dbName, tableName);
- AllocateTableWriteIdsResponse writeIds = client.allocate_table_write_ids(rqst);
- return writeIds.getTxnToWriteIds();
+ AllocateTableWriteIdsRequest rqst = new AllocateTableWriteIdsRequest(dbName, tableName);
+ rqst.setTxnIds(txnIds);
+ return allocateTableWriteIdsBatchIntr(rqst);
+ }
+
+ @Override
+ public List<TxnToWriteId> replAllocateTableWriteIdsBatch(String dbName, String tableName,
+ String replPolicy, List<TxnToWriteId> srcTxnToWriteIdList) throws TException {
+ AllocateTableWriteIdsRequest rqst = new AllocateTableWriteIdsRequest(dbName, tableName);
+ rqst.setReplPolicy(replPolicy);
+ rqst.setSrcTxnToWriteIdList(srcTxnToWriteIdList);
+ return allocateTableWriteIdsBatchIntr(rqst);
+ }
+
+ private List<TxnToWriteId> allocateTableWriteIdsBatchIntr(AllocateTableWriteIdsRequest rqst) throws TException {
+ return client.allocate_table_write_ids(rqst).getTxnToWriteIds();
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
index 98674cf..27f8775 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
@@ -27,7 +27,6 @@ import java.util.Map.Entry;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.common.ValidTxnList;
import org.apache.hadoop.hive.common.ValidTxnWriteIdList;
@@ -36,7 +35,6 @@ import org.apache.hadoop.hive.common.classification.RetrySemantics;
import org.apache.hadoop.hive.metastore.annotation.NoReconnect;
import org.apache.hadoop.hive.metastore.api.AggrStats;
import org.apache.hadoop.hive.metastore.api.AlreadyExistsException;
-import org.apache.hadoop.hive.metastore.api.BasicTxnInfo;
import org.apache.hadoop.hive.metastore.api.CheckConstraintsRequest;
import org.apache.hadoop.hive.metastore.api.Catalog;
import org.apache.hadoop.hive.metastore.api.CmRecycleRequest;
@@ -127,7 +125,6 @@ import org.apache.hadoop.hive.metastore.api.WMResourcePlan;
import org.apache.hadoop.hive.metastore.api.WMTrigger;
import org.apache.hadoop.hive.metastore.api.WMValidateResourcePlanResponse;
import org.apache.hadoop.hive.metastore.partition.spec.PartitionSpecProxy;
-import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
import org.apache.hadoop.hive.metastore.utils.ObjectPair;
import org.apache.thrift.TException;
@@ -2884,6 +2881,16 @@ public interface IMetaStoreClient {
List<TxnToWriteId> allocateTableWriteIdsBatch(List<Long> txnIds, String dbName, String tableName) throws TException;
/**
+ * Allocate a per table write ID and associate it with the given transaction. Used by replication load task.
+ * @param dbName name of DB in which the table belongs.
+ * @param tableName table to which the write ID to be allocated
+ * @param replPolicy Used by replication task to identify the source cluster.
+ * @param srcTxnToWriteIdList List of txn to write id map sent from the source cluster.
+ * @throws TException
+ */
+ List<TxnToWriteId> replAllocateTableWriteIdsBatch(String dbName, String tableName, String replPolicy,
+ List<TxnToWriteId> srcTxnToWriteIdList) throws TException;
+ /**
* Show the list of currently open transactions. This is for use by "show transactions" in the
* grammar, not for applications that want to find a list of current transactions to work with.
* Those wishing the latter should call {@link #getValidTxns()}.
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreEventListener.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreEventListener.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreEventListener.java
index 7b3a80c..92505af 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreEventListener.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreEventListener.java
@@ -53,6 +53,9 @@ import org.apache.hadoop.hive.metastore.events.LoadPartitionDoneEvent;
import org.apache.hadoop.hive.metastore.events.OpenTxnEvent;
import org.apache.hadoop.hive.metastore.events.CommitTxnEvent;
import org.apache.hadoop.hive.metastore.events.AbortTxnEvent;
+import org.apache.hadoop.hive.metastore.events.AllocWriteIdEvent;
+import org.apache.hadoop.hive.metastore.tools.SQLGenerator;
+import java.sql.Connection;
/**
* This abstract class needs to be extended to provide implementation of actions that needs
@@ -234,26 +237,45 @@ public abstract class MetaStoreEventListener implements Configurable {
/**
* This will be called when a new transaction is started.
- * @param openTxnEvent
+ * @param openTxnEvent event to be processed
+ * @param dbConn jdbc connection to remote meta store db.
+ * @param sqlGenerator helper class to generate db specific sql string.
* @throws MetaException
*/
- public void onOpenTxn(OpenTxnEvent openTxnEvent) throws MetaException {
+ public void onOpenTxn(OpenTxnEvent openTxnEvent, Connection dbConn, SQLGenerator sqlGenerator) throws MetaException {
}
/**
* This will be called to commit a transaction.
- * @param commitTxnEvent
+ * @param commitTxnEvent event to be processed
+ * @param dbConn jdbc connection to remote meta store db.
+ * @param sqlGenerator helper class to generate db specific sql string.
* @throws MetaException
*/
- public void onCommitTxn(CommitTxnEvent commitTxnEvent) throws MetaException {
+ public void onCommitTxn(CommitTxnEvent commitTxnEvent, Connection dbConn, SQLGenerator sqlGenerator) throws
+ MetaException {
}
/**
* This will be called to abort a transaction.
- * @param abortTxnEvent
+ * @param abortTxnEvent event to be processed
+ * @param dbConn jdbc connection to remote meta store db.
+ * @param sqlGenerator helper class to generate db specific sql string.
* @throws MetaException
*/
- public void onAbortTxn(AbortTxnEvent abortTxnEvent) throws MetaException {
+ public void onAbortTxn(AbortTxnEvent abortTxnEvent, Connection dbConn, SQLGenerator sqlGenerator)
+ throws MetaException {
+ }
+
+ /**
+ * This will be called to alloc a new write id.
+ * @param allocWriteIdEvent event to be processed
+ * @param dbConn jdbc connection to remote meta store db.
+ * @param sqlGenerator helper class to generate db specific sql string.
+ * @throws MetaException
+ */
+ public void onAllocWriteId(AllocWriteIdEvent allocWriteIdEvent, Connection dbConn, SQLGenerator sqlGenerator)
+ throws MetaException {
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreListenerNotifier.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreListenerNotifier.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreListenerNotifier.java
index e9bbfdc..b2856e2 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreListenerNotifier.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreListenerNotifier.java
@@ -52,10 +52,11 @@ import org.apache.hadoop.hive.metastore.events.ListenerEvent;
import org.apache.hadoop.hive.metastore.events.OpenTxnEvent;
import org.apache.hadoop.hive.metastore.events.CommitTxnEvent;
import org.apache.hadoop.hive.metastore.events.AbortTxnEvent;
-
+import org.apache.hadoop.hive.metastore.events.AllocWriteIdEvent;
+import org.apache.hadoop.hive.metastore.tools.SQLGenerator;
+import java.sql.Connection;
import java.util.List;
import java.util.Map;
-
import static org.apache.hadoop.hive.metastore.MetaStoreEventListenerConstants.HIVE_METASTORE_TRANSACTION_ACTIVE;
import static org.apache.hadoop.hive.metastore.messaging.EventMessage.EventType;
@@ -209,29 +210,37 @@ public class MetaStoreListenerNotifier {
(listener, event) -> listener.onCreateCatalog((CreateCatalogEvent)event))
.put(EventType.DROP_CATALOG,
(listener, event) -> listener.onDropCatalog((DropCatalogEvent)event))
- .put(EventType.OPEN_TXN, new EventNotifier() {
- @Override
- public void notify(MetaStoreEventListener listener, ListenerEvent event) throws MetaException {
- listener.onOpenTxn((OpenTxnEvent)event);
- }
- })
- .put(EventType.COMMIT_TXN, new EventNotifier() {
- @Override
- public void notify(MetaStoreEventListener listener, ListenerEvent event)
- throws MetaException {
- listener.onCommitTxn((CommitTxnEvent) event);
- }
- })
- .put(EventType.ABORT_TXN, new EventNotifier() {
- @Override
- public void notify(MetaStoreEventListener listener, ListenerEvent event)
- throws MetaException {
- listener.onAbortTxn((AbortTxnEvent) event);
- }
- })
+ .put(EventType.OPEN_TXN,
+ (listener, event) -> listener.onOpenTxn((OpenTxnEvent) event, null, null))
+ .put(EventType.COMMIT_TXN,
+ (listener, event) -> listener.onCommitTxn((CommitTxnEvent) event, null, null))
+ .put(EventType.ABORT_TXN,
+ (listener, event) -> listener.onAbortTxn((AbortTxnEvent) event, null, null))
+ .put(EventType.ALLOC_WRITE_ID,
+ (listener, event) -> listener.onAllocWriteId((AllocWriteIdEvent) event, null, null))
.build()
);
+
+ private interface TxnEventNotifier {
+ void notify(MetaStoreEventListener listener, ListenerEvent event, Connection dbConn, SQLGenerator sqlGenerator)
+ throws MetaException;
+ }
+
+ private static Map<EventType, TxnEventNotifier> txnNotificationEvents = Maps.newHashMap(
+ ImmutableMap.<EventType, TxnEventNotifier>builder()
+ .put(EventType.OPEN_TXN,
+ (listener, event, dbConn, sqlGenerator) -> listener.onOpenTxn((OpenTxnEvent) event, dbConn, sqlGenerator))
+ .put(EventType.COMMIT_TXN,
+ (listener, event, dbConn, sqlGenerator) -> listener.onCommitTxn((CommitTxnEvent) event, dbConn, sqlGenerator))
+ .put(EventType.ABORT_TXN,
+ (listener, event, dbConn, sqlGenerator) -> listener.onAbortTxn((AbortTxnEvent) event, dbConn, sqlGenerator))
+ .put(EventType.ALLOC_WRITE_ID,
+ (listener, event, dbConn, sqlGenerator) ->
+ listener.onAllocWriteId((AllocWriteIdEvent) event, dbConn, sqlGenerator))
+ .build()
+ );
+
/**
* Notify a list of listeners about a specific metastore event. Each listener notified might update
* the (ListenerEvent) event by setting a parameter key/value pair. These updated parameters will
@@ -262,6 +271,38 @@ public class MetaStoreListenerNotifier {
}
/**
+ * Notify a list of listeners about a specific metastore event to be executed within a txn. Each listener notified
+ * might update the (ListenerEvent) event by setting a parameter key/value pair. These updated parameters will
+ * be returned to the caller.
+ *
+ * @param listeners List of MetaStoreEventListener listeners.
+ * @param eventType Type of the notification event.
+ * @param event The ListenerEvent with information about the event.
+ * @param dbConn The JDBC connection to the remote meta store db.
+ * @param sqlGenerator The helper class to generate db specific SQL string.
+ * @return A list of key/value pair parameters that the listeners set. The returned object will return an empty
+ * map if no parameters were updated or if no listeners were notified.
+ * @throws MetaException If an error occurred while calling the listeners.
+ */
+ public static Map<String, String> notifyEventWithDirectSql(List<? extends MetaStoreEventListener> listeners,
+ EventType eventType,
+ ListenerEvent event,
+ Connection dbConn, SQLGenerator sqlGenerator) throws MetaException {
+
+ Preconditions.checkNotNull(listeners, "Listeners must not be null.");
+ Preconditions.checkNotNull(event, "The event must not be null.");
+
+ for (MetaStoreEventListener listener : listeners) {
+ txnNotificationEvents.get(eventType).notify(listener, event, dbConn, sqlGenerator);
+ }
+
+ // Each listener called above might set a different parameter on the event.
+ // This write permission is allowed on the listener side to avoid breaking compatibility if we change the API
+ // method calls.
+ return event.getParameters();
+ }
+
+ /**
* Notify a list of listeners about a specific metastore event. Each listener notified might update
* the (ListenerEvent) event by setting a parameter key/value pair. These updated parameters will
* be returned to the caller.
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/AbortTxnEvent.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/AbortTxnEvent.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/AbortTxnEvent.java
index 062e719..fe4b974 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/AbortTxnEvent.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/AbortTxnEvent.java
@@ -21,8 +21,6 @@ package org.apache.hadoop.hive.metastore.events;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.hive.metastore.IHMSHandler;
-import org.apache.hadoop.hive.metastore.tools.SQLGenerator;
-import java.sql.Connection;
/**
* AbortTxnEvent
@@ -33,8 +31,6 @@ import java.sql.Connection;
public class AbortTxnEvent extends ListenerEvent {
private final Long txnId;
- Connection connection;
- SQLGenerator sqlGenerator;
/**
*
@@ -44,20 +40,6 @@ public class AbortTxnEvent extends ListenerEvent {
public AbortTxnEvent(Long transactionId, IHMSHandler handler) {
super(true, handler);
txnId = transactionId;
- connection = null;
- sqlGenerator = null;
- }
-
- /**
- * @param transactionId Unique identification for the transaction just got aborted.
- * @param connection connection to execute direct SQL statement within same transaction
- * @param sqlGenerator generates db specific SQL query
- */
- public AbortTxnEvent(Long transactionId, Connection connection, SQLGenerator sqlGenerator) {
- super(true, null);
- this.txnId = transactionId;
- this.connection = connection;
- this.sqlGenerator = sqlGenerator;
}
/**
@@ -66,18 +48,4 @@ public class AbortTxnEvent extends ListenerEvent {
public Long getTxnId() {
return txnId;
}
-
- /**
- * @return Connection connection - used only by DbNotificationListener
- */
- public Connection getConnection() {
- return connection;
- }
-
- /**
- * @return SQLGenerator sqlGenerator - used only by DbNotificationListener
- */
- public SQLGenerator getSqlGenerator() {
- return sqlGenerator;
- }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/AllocWriteIdEvent.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/AllocWriteIdEvent.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/AllocWriteIdEvent.java
new file mode 100644
index 0000000..84a9a4e
--- /dev/null
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/AllocWriteIdEvent.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.metastore.events;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hive.metastore.IHMSHandler;
+import org.apache.hadoop.hive.metastore.api.TxnToWriteId;
+import java.util.List;
+
+/**
+ * AllocWriteIdEvent.
+ * Event for allocating write id.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public class AllocWriteIdEvent extends ListenerEvent {
+
+ private final List<TxnToWriteId> txnToWriteIdList;
+ private final String tableName;
+ private final String dbName;
+
+ public AllocWriteIdEvent(List<TxnToWriteId> txnToWriteIdList, String dbName, String tableName, IHMSHandler handler) {
+ super(true, handler);
+ this.txnToWriteIdList = txnToWriteIdList;
+ this.tableName = tableName;
+ this.dbName = dbName;
+ }
+
+ public List<TxnToWriteId> getTxnToWriteIdList() {
+ return txnToWriteIdList;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ public String getDbName() {
+ return dbName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/CommitTxnEvent.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/CommitTxnEvent.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/CommitTxnEvent.java
index 7262e6b..ba382cd 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/CommitTxnEvent.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/CommitTxnEvent.java
@@ -21,8 +21,6 @@ package org.apache.hadoop.hive.metastore.events;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.hive.metastore.IHMSHandler;
-import org.apache.hadoop.hive.metastore.tools.SQLGenerator;
-import java.sql.Connection;
/**
* CommitTxnEvent
@@ -33,8 +31,6 @@ import java.sql.Connection;
public class CommitTxnEvent extends ListenerEvent {
private final Long txnId;
- Connection connection;
- SQLGenerator sqlGenerator;
/**
*
@@ -44,20 +40,6 @@ public class CommitTxnEvent extends ListenerEvent {
public CommitTxnEvent(Long transactionId, IHMSHandler handler) {
super(true, handler);
this.txnId = transactionId;
- this.connection = null;
- this.sqlGenerator = null;
- }
-
- /**
- * @param transactionId Unique identification for the transaction just got committed.
- * @param connection connection to execute direct SQL statement within same transaction
- * @param sqlGenerator generates db specific SQL query
- */
- public CommitTxnEvent(Long transactionId, Connection connection, SQLGenerator sqlGenerator) {
- super(true, null);
- this.txnId = transactionId;
- this.connection = connection;
- this.sqlGenerator = sqlGenerator;
}
/**
@@ -66,18 +48,4 @@ public class CommitTxnEvent extends ListenerEvent {
public Long getTxnId() {
return txnId;
}
-
- /**
- * @return Connection connection - used only by DbNotificationListener
- */
- public Connection getConnection() {
- return connection;
- }
-
- /**
- * @return SQLGenerator sqlGenerator - used only by DbNotificationListener
- */
- public SQLGenerator getSqlGenerator() {
- return sqlGenerator;
- }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/ListenerEvent.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/ListenerEvent.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/ListenerEvent.java
index b542afc..93f1c2b 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/ListenerEvent.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/ListenerEvent.java
@@ -23,9 +23,7 @@ import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.hive.metastore.HiveMetaStore;
import org.apache.hadoop.hive.metastore.IHMSHandler;
import org.apache.hadoop.hive.metastore.api.EnvironmentContext;
-import org.apache.hadoop.hive.metastore.tools.SQLGenerator;
import javax.annotation.concurrent.NotThreadSafe;
-import java.sql.Connection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -163,20 +161,6 @@ public abstract class ListenerEvent {
}
/**
- * Used by ACID/transaction related events for generating direct SQL in DBNotificationListener.
- */
- public Connection getConnection() {
- return null;
- }
-
- /**
- * Used by ACID/transaction related events for generating direct SQL in DBNotificationListener.
- */
- public SQLGenerator getSqlGenerator() {
- return null;
- }
-
- /**
* Put a parameter to the listener event only if the parameter is absent.
*
* Overridden parameters is not allowed, and an exception may be thrown to avoid a mis-configuration
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/OpenTxnEvent.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/OpenTxnEvent.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/OpenTxnEvent.java
index 088a6a1..547c43e 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/OpenTxnEvent.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/events/OpenTxnEvent.java
@@ -22,8 +22,6 @@ import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.hive.metastore.IHMSHandler;
import com.google.common.collect.Lists;
-import org.apache.hadoop.hive.metastore.tools.SQLGenerator;
-import java.sql.Connection;
import java.util.List;
/**
@@ -34,8 +32,6 @@ import java.util.List;
@InterfaceStability.Stable
public class OpenTxnEvent extends ListenerEvent {
private List<Long> txnIds;
- Connection connection;
- SQLGenerator sqlGenerator;
/**
* @param txnIds List of unique identification for the transaction just opened.
@@ -44,20 +40,6 @@ public class OpenTxnEvent extends ListenerEvent {
public OpenTxnEvent(List<Long> txnIds, IHMSHandler handler) {
super(true, handler);
this.txnIds = Lists.newArrayList(txnIds);
- this.connection = null;
- this.sqlGenerator = null;
- }
-
- /**
- * @param txnIds List of unique identification for the transaction just opened.
- * @param connection connection to execute direct SQL statement within same transaction
- * @param sqlGenerator generates db specific SQL query
- */
- public OpenTxnEvent(List<Long> txnIds, Connection connection, SQLGenerator sqlGenerator) {
- super(true, null);
- this.txnIds = Lists.newArrayList(txnIds);
- this.connection = connection;
- this.sqlGenerator = sqlGenerator;
}
/**
@@ -66,18 +48,4 @@ public class OpenTxnEvent extends ListenerEvent {
public List<Long> getTxnIds() {
return txnIds;
}
-
- /**
- * @return Connection connection - used only by DbNotificationListener
- */
- public Connection getConnection() {
- return connection;
- }
-
- /**
- * @return SQLGenerator sqlGenerator - used only by DbNotificationListener
- */
- public SQLGenerator getSqlGenerator() {
- return sqlGenerator;
- }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/AllocWriteIdMessage.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/AllocWriteIdMessage.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/AllocWriteIdMessage.java
new file mode 100644
index 0000000..0766d02
--- /dev/null
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/AllocWriteIdMessage.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hadoop.hive.metastore.messaging;
+
+import org.apache.hadoop.hive.metastore.api.TxnToWriteId;
+
+import java.util.List;
+
+/**
+ * HCat message sent when an alloc write id is done.
+ */
+public abstract class AllocWriteIdMessage extends EventMessage {
+ protected AllocWriteIdMessage() {
+ super(EventType.ALLOC_WRITE_ID);
+ }
+
+ public abstract String getTableName();
+
+ public abstract List<TxnToWriteId> getTxnToWriteIdList();
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/EventMessage.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/EventMessage.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/EventMessage.java
index 5137c86..ffbce1d 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/EventMessage.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/EventMessage.java
@@ -58,7 +58,8 @@ public abstract class EventMessage {
DROP_CATALOG(MessageFactory.DROP_CATALOG_EVENT),
OPEN_TXN(MessageFactory.OPEN_TXN_EVENT),
COMMIT_TXN(MessageFactory.COMMIT_TXN_EVENT),
- ABORT_TXN(MessageFactory.ABORT_TXN_EVENT);
+ ABORT_TXN(MessageFactory.ABORT_TXN_EVENT),
+ ALLOC_WRITE_ID(MessageFactory.ALLOC_WRITE_ID_EVENT);
private String typeString;
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageDeserializer.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageDeserializer.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageDeserializer.java
index 6583cc7..ca33579 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageDeserializer.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageDeserializer.java
@@ -181,6 +181,11 @@ public abstract class MessageDeserializer {
*/
public abstract AbortTxnMessage getAbortTxnMessage(String messageBody);
+ /*
+ * Method to de-serialize AllocWriteIdMessage instance.
+ */
+ public abstract AllocWriteIdMessage getAllocWriteIdMessage(String messageBody);
+
// Protection against construction.
protected MessageDeserializer() {}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageFactory.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageFactory.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageFactory.java
index dc4420e..75ca6ec 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageFactory.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageFactory.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.hive.metastore.api.SQLNotNullConstraint;
import org.apache.hadoop.hive.metastore.api.SQLPrimaryKey;
import org.apache.hadoop.hive.metastore.api.SQLUniqueConstraint;
import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.api.TxnToWriteId;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
import org.apache.hadoop.hive.metastore.utils.JavaUtils;
@@ -71,6 +72,7 @@ public abstract class MessageFactory {
public static final String OPEN_TXN_EVENT = "OPEN_TXN";
public static final String COMMIT_TXN_EVENT = "COMMIT_TXN";
public static final String ABORT_TXN_EVENT = "ABORT_TXN";
+ public static final String ALLOC_WRITE_ID_EVENT = "ALLOC_WRITE_ID_EVENT";
private static MessageFactory instance = null;
@@ -265,6 +267,17 @@ public abstract class MessageFactory {
*/
public abstract AbortTxnMessage buildAbortTxnMessage(Long txnId);
+ /**
+ * Factory method for building alloc write id message
+ *
+ * @param txnToWriteIdList List of Txn Ids and write id map
+ * @param dbName db for which write ids to be allocated
+ * @param tableName table for which write ids to be allocated
+ * @return instance of AllocWriteIdMessage
+ */
+ public abstract AllocWriteIdMessage buildAllocWriteIdMessage(List<TxnToWriteId> txnToWriteIdList, String dbName,
+ String tableName);
+
/***
* Factory method for building add primary key message
*
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/event/filters/DatabaseAndTableFilter.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/event/filters/DatabaseAndTableFilter.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/event/filters/DatabaseAndTableFilter.java
index 09292a3..fdb6942 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/event/filters/DatabaseAndTableFilter.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/event/filters/DatabaseAndTableFilter.java
@@ -43,8 +43,7 @@ public class DatabaseAndTableFilter extends BasicFilter {
private boolean isTxnRelatedEvent(final NotificationEvent event) {
return ((event.getEventType().equals(MessageFactory.OPEN_TXN_EVENT)) ||
(event.getEventType().equals(MessageFactory.COMMIT_TXN_EVENT)) ||
- (event.getEventType().equals(MessageFactory.ABORT_TXN_EVENT))
- );
+ (event.getEventType().equals(MessageFactory.ABORT_TXN_EVENT)));
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONAllocWriteIdMessage.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONAllocWriteIdMessage.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONAllocWriteIdMessage.java
new file mode 100644
index 0000000..d4aba11
--- /dev/null
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONAllocWriteIdMessage.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hadoop.hive.metastore.messaging.json;
+
+import org.apache.hadoop.hive.metastore.api.TxnToWriteId;
+import org.apache.hadoop.hive.metastore.messaging.AllocWriteIdMessage;
+import org.codehaus.jackson.annotate.JsonProperty;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * JSON implementation of AllocWriteId
+ */
+public class JSONAllocWriteIdMessage extends AllocWriteIdMessage {
+
+ @JsonProperty
+ private String server, servicePrincipal, dbName, tableName;
+
+ @JsonProperty
+ private List<Long> txnIdList, writeIdList;
+
+ @JsonProperty
+ private Long timestamp;
+
+ private List<TxnToWriteId> txnToWriteIdList;
+
+ /**
+ * Default constructor, needed for Jackson.
+ */
+ public JSONAllocWriteIdMessage() {
+ }
+
+ public JSONAllocWriteIdMessage(String server, String servicePrincipal,
+ List<TxnToWriteId> txnToWriteIdList, String dbName, String tableName, Long timestamp) {
+ this.server = server;
+ this.servicePrincipal = servicePrincipal;
+ this.timestamp = timestamp;
+ this.txnIdList = new ArrayList<>();
+ this.writeIdList = new ArrayList<>();
+ for (TxnToWriteId txnToWriteId : txnToWriteIdList) {
+ this.txnIdList.add(txnToWriteId.getTxnId());
+ this.writeIdList.add(txnToWriteId.getWriteId());
+ }
+ this.tableName = tableName;
+ this.dbName = dbName;
+ this.txnToWriteIdList = txnToWriteIdList;
+ }
+
+ @Override
+ public String getServer() {
+ return server;
+ }
+
+ @Override
+ public String getServicePrincipal() {
+ return servicePrincipal;
+ }
+
+ @Override
+ public String getDB() {
+ return dbName;
+ }
+
+ @Override
+ public Long getTimestamp() {
+ return timestamp;
+ }
+
+ @Override
+ public String getTableName() {
+ return tableName;
+ }
+
+ @Override
+ public List<TxnToWriteId> getTxnToWriteIdList() {
+ // after deserialization, need to recreate the txnToWriteIdList as its not under JsonProperty.
+ if (txnToWriteIdList == null) {
+ txnToWriteIdList = new ArrayList<>();
+ for (int idx = 0; idx < txnIdList.size(); idx++) {
+ txnToWriteIdList.add(new TxnToWriteId(txnIdList.get(idx), writeIdList.get(idx)));
+ }
+ }
+ return txnToWriteIdList;
+ }
+
+ @Override
+ public String toString() {
+ try {
+ return JSONMessageDeserializer.mapper.writeValueAsString(this);
+ } catch (Exception exception) {
+ throw new IllegalArgumentException("Could not serialize: ", exception);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONMessageDeserializer.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONMessageDeserializer.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONMessageDeserializer.java
index d019ec1..f54e24d 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONMessageDeserializer.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONMessageDeserializer.java
@@ -40,6 +40,7 @@ import org.apache.hadoop.hive.metastore.messaging.MessageDeserializer;
import org.apache.hadoop.hive.metastore.messaging.OpenTxnMessage;
import org.apache.hadoop.hive.metastore.messaging.CommitTxnMessage;
import org.apache.hadoop.hive.metastore.messaging.AbortTxnMessage;
+import org.apache.hadoop.hive.metastore.messaging.AllocWriteIdMessage;
import org.codehaus.jackson.map.DeserializationConfig;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.map.SerializationConfig;
@@ -250,4 +251,12 @@ public class JSONMessageDeserializer extends MessageDeserializer {
throw new IllegalArgumentException("Could not construct AbortTxnMessage", e);
}
}
+
+ public AllocWriteIdMessage getAllocWriteIdMessage(String messageBody) {
+ try {
+ return mapper.readValue(messageBody, JSONAllocWriteIdMessage.class);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Could not construct AllocWriteIdMessage", e);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/fa9e743e/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONMessageFactory.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONMessageFactory.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONMessageFactory.java
index 65a6f78..f0c5f4f 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONMessageFactory.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONMessageFactory.java
@@ -38,6 +38,7 @@ import org.apache.hadoop.hive.metastore.api.SQLNotNullConstraint;
import org.apache.hadoop.hive.metastore.api.SQLPrimaryKey;
import org.apache.hadoop.hive.metastore.api.SQLUniqueConstraint;
import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.api.TxnToWriteId;
import org.apache.hadoop.hive.metastore.messaging.AddForeignKeyMessage;
import org.apache.hadoop.hive.metastore.messaging.AddNotNullConstraintMessage;
import org.apache.hadoop.hive.metastore.messaging.AddPartitionMessage;
@@ -63,6 +64,7 @@ import org.apache.hadoop.hive.metastore.messaging.PartitionFiles;
import org.apache.hadoop.hive.metastore.messaging.OpenTxnMessage;
import org.apache.hadoop.hive.metastore.messaging.CommitTxnMessage;
import org.apache.hadoop.hive.metastore.messaging.AbortTxnMessage;
+import org.apache.hadoop.hive.metastore.messaging.AllocWriteIdMessage;
import org.apache.thrift.TBase;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
@@ -221,6 +223,11 @@ public class JSONMessageFactory extends MessageFactory {
return new JSONAbortTxnMessage(MS_SERVER_URL, MS_SERVICE_PRINCIPAL, txnId, now());
}
+ public AllocWriteIdMessage buildAllocWriteIdMessage(List<TxnToWriteId> txnToWriteIdList,
+ String dbName, String tableName) {
+ return new JSONAllocWriteIdMessage(MS_SERVER_URL, MS_SERVICE_PRINCIPAL, txnToWriteIdList, dbName, tableName, now());
+ }
+
private long now() {
return System.currentTimeMillis() / 1000;
}