2020import java .util .concurrent .ExecutionException ;
2121import java .util .concurrent .TimeUnit ;
2222
23+ import com .amazonaws .services .dynamodbv2 .model .CreateTableRequest ;
24+ import com .amazonaws .services .dynamodbv2 .model .DeleteItemRequest ;
25+ import com .amazonaws .services .dynamodbv2 .model .GetItemRequest ;
26+ import com .amazonaws .services .dynamodbv2 .model .ProvisionedThroughput ;
27+ import com .amazonaws .services .dynamodbv2 .model .QueryRequest ;
28+ import com .amazonaws .services .dynamodbv2 .model .ReturnConsumedCapacity ;
29+ import com .amazonaws .services .dynamodbv2 .model .ScanRequest ;
30+ import com .amazonaws .services .dynamodbv2 .model .UpdateItemRequest ;
31+ import lombok .Getter ;
2332import org .apache .commons .codec .binary .Hex ;
33+ import org .apache .commons .lang .builder .EqualsBuilder ;
2434import org .apache .commons .lang3 .tuple .Pair ;
2535import org .janusgraph .diskstorage .BackendException ;
2636import org .janusgraph .diskstorage .Entry ;
3141import org .janusgraph .diskstorage .keycolumnvalue .StoreTransaction ;
3242import org .janusgraph .diskstorage .locking .TemporaryLockingException ;
3343
34- import com .amazonaws .services .dynamodbv2 .model .CreateTableRequest ;
35- import com .amazonaws .services .dynamodbv2 .model .GetItemRequest ;
36- import com .amazonaws .services .dynamodbv2 .model .ProvisionedThroughput ;
37- import com .amazonaws .services .dynamodbv2 .model .QueryRequest ;
38- import com .amazonaws .services .dynamodbv2 .model .ScanRequest ;
39- import com .amazonaws .services .dynamodbv2 .model .UpdateItemRequest ;
4044import com .google .common .cache .Cache ;
4145import com .google .common .cache .CacheBuilder ;
4246import com .google .common .cache .RemovalListener ;
5761@ Slf4j
5862public abstract class AbstractDynamoDbStore implements AwsStore {
5963 protected final Client client ;
60- protected final String tableName ;
64+ @ Getter
65+ private final String tableName ;
6166 private final DynamoDBStoreManager manager ;
62- private final String storeName ;
67+ @ Getter
68+ private final String name ;
6369 private final boolean forceConsistentRead ;
6470 private final Cache <Pair <StaticBuffer , StaticBuffer >, DynamoDbStoreTransaction > keyColumnLocalLocks ;
6571
66- private static final class ReportingRemovalListener implements RemovalListener <Pair <StaticBuffer , StaticBuffer >, DynamoDbStoreTransaction > {
67- private static final ReportingRemovalListener INSTANCE = new ReportingRemovalListener ();
68- private static ReportingRemovalListener theInstance () {
69- return INSTANCE ;
70- }
71- private ReportingRemovalListener () { }
72+ private enum ReportingRemovalListener implements RemovalListener <Pair <StaticBuffer , StaticBuffer >, DynamoDbStoreTransaction > {
73+ INSTANCE ;
7274
7375 @ Override
7476 public void onRemoval (final RemovalNotification <Pair <StaticBuffer , StaticBuffer >, DynamoDbStoreTransaction > notice ) {
7577 log .trace ("Expiring {} in tx {} because of {}" , notice .getKey ().toString (), notice .getValue ().toString (), notice .getCause ());
7678 }
7779 }
7880
79- protected CreateTableRequest createTableRequest () {
80- return new CreateTableRequest ()
81- .withTableName (tableName )
82- .withProvisionedThroughput (new ProvisionedThroughput (client .readCapacity (tableName ),
83- client .writeCapacity (tableName )));
84- }
85-
8681 protected void mutateOneKey (final StaticBuffer key , final KCVMutation mutation , final StoreTransaction txh ) throws BackendException {
87- manager .mutateMany (Collections .singletonMap (storeName , Collections .singletonMap (key , mutation )), txh );
88- }
89-
90- @ Override
91- public String getName () {
92- return storeName ;
82+ manager .mutateMany (Collections .singletonMap (name , Collections .singletonMap (key , mutation )), txh );
9383 }
9484
9585 protected UpdateItemRequest createUpdateItemRequest () {
96- return new UpdateItemRequest ().withTableName (tableName );
86+ return new UpdateItemRequest ()
87+ .withTableName (tableName )
88+ .withReturnConsumedCapacity (ReturnConsumedCapacity .TOTAL );
9789 }
9890
9991 protected GetItemRequest createGetItemRequest () {
100- return new GetItemRequest ().withTableName (tableName ).withConsistentRead (forceConsistentRead );
92+ return new GetItemRequest ()
93+ .withTableName (tableName )
94+ .withConsistentRead (forceConsistentRead )
95+ .withReturnConsumedCapacity (ReturnConsumedCapacity .TOTAL );
96+ }
97+
98+ protected DeleteItemRequest createDeleteItemRequest () {
99+ return new DeleteItemRequest ()
100+ .withTableName (tableName )
101+ .withReturnConsumedCapacity (ReturnConsumedCapacity .TOTAL );
101102 }
102103
103104 protected QueryRequest createQueryRequest () {
104- return new QueryRequest ().withTableName (tableName ).withConsistentRead (forceConsistentRead );
105+ return new QueryRequest ()
106+ .withTableName (tableName )
107+ .withConsistentRead (forceConsistentRead )
108+ .withReturnConsumedCapacity (ReturnConsumedCapacity .TOTAL );
105109 }
106110 protected ScanRequest createScanRequest () {
107- return new ScanRequest ().withTableName (tableName ).withConsistentRead (forceConsistentRead );
111+ return new ScanRequest ().withTableName (tableName )
112+ .withConsistentRead (forceConsistentRead )
113+ .withLimit (client .scanLimit (tableName ))
114+ .withReturnConsumedCapacity (ReturnConsumedCapacity .TOTAL );
108115 }
109116 AbstractDynamoDbStore (final DynamoDBStoreManager manager , final String prefix , final String storeName ) {
110117 this .manager = manager ;
111118 this .client = this .manager .getClient ();
112- this .storeName = storeName ;
119+ this .name = storeName ;
113120 this .tableName = prefix + "_" + storeName ;
114121 this .forceConsistentRead = client .isForceConsistentRead ();
115122
116123 final CacheBuilder <Pair <StaticBuffer , StaticBuffer >, DynamoDbStoreTransaction > builder = CacheBuilder .newBuilder ().concurrencyLevel (client .getDelegate ().getMaxConcurrentUsers ())
117124 .expireAfterWrite (manager .getLockExpiresDuration ().toMillis (), TimeUnit .MILLISECONDS )
118- .removalListener (ReportingRemovalListener .theInstance () );
125+ .removalListener (ReportingRemovalListener .INSTANCE );
119126 this .keyColumnLocalLocks = builder .build ();
120127 }
121128
122129 /**
123130 * Creates the schemata for the DynamoDB table or tables each store requires.
131+ * Implementations should override and reuse this logic
124132 * @return a create table request appropriate for the schema of the selected implementation.
125133 */
126- public abstract CreateTableRequest getTableSchema ();
134+ public CreateTableRequest getTableSchema () {
135+ return new CreateTableRequest ()
136+ .withTableName (tableName )
137+ .withProvisionedThroughput (new ProvisionedThroughput (client .readCapacity (tableName ),
138+ client .writeCapacity (tableName )));
139+ }
127140
128141 @ Override
129142 public final void ensureStore () throws BackendException {
@@ -133,7 +146,7 @@ public final void ensureStore() throws BackendException {
133146
134147 @ Override
135148 public final void deleteStore () throws BackendException {
136- log .debug ("Entering deleteStore name:{}" , storeName );
149+ log .debug ("Entering deleteStore name:{}" , name );
137150 client .getDelegate ().deleteTable (getTableSchema ().getTableName ());
138151 //block until the tables are actually deleted
139152 client .getDelegate ().ensureTableDeleted (getTableSchema ().getTableName ());
@@ -179,11 +192,6 @@ public void close() throws BackendException {
179192 log .debug ("Closing table:{}" , tableName );
180193 }
181194
182- @ Override
183- public String getTableName () {
184- return tableName ;
185- }
186-
187195 protected String encodeKeyForLog (final StaticBuffer key ) {
188196 if (null == key ) {
189197 return "" ;
@@ -209,6 +217,31 @@ String encodeForLog(final List<?> columns) {
209217 return result .append ("]" ).toString ();
210218 }
211219
220+ @ Override
221+ public int hashCode () {
222+ return tableName .hashCode ();
223+ }
224+
225+ @ Override
226+ public boolean equals (final Object obj ) {
227+ if (obj == null ) {
228+ return false ;
229+ }
230+ if (obj == this ) {
231+ return true ;
232+ }
233+ if (obj .getClass () != getClass ()) {
234+ return false ;
235+ }
236+ final AbstractDynamoDbStore rhs = (AbstractDynamoDbStore ) obj ;
237+ return new EqualsBuilder ().append (tableName , rhs .tableName ).isEquals ();
238+ }
239+
240+ @ Override
241+ public String toString () {
242+ return this .getClass ().getName () + ":" + getTableName ();
243+ }
244+
212245 protected String encodeForLog (final SliceQuery query ) {
213246 return "slice[rk:" + encodeKeyForLog (query .getSliceStart ()) + " -> " + encodeKeyForLog (query .getSliceEnd ()) + " limit:" + query .getLimit () + "]" ;
214247 }
0 commit comments