Skip to content

Commit ccb5cef

Browse files
brfrn169brfrn169
committed
Support batch operation in Transaction API
1 parent de34b99 commit ccb5cef

File tree

42 files changed

+2972
-151
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

42 files changed

+2972
-151
lines changed

‎core/src/main/java/com/scalar/db/api/CrudOperable.java‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -156,6 +156,17 @@ public interface CrudOperable<E extends TransactionException> {
156156
*/
157157
void mutate(List<? extends Mutation> mutations) throws E;
158158

159+
/**
160+
* Executes multiple operations in a batch through a transaction with the specified list of {@link
161+
* Operation} commands and returns a list of {@link BatchResult} that contains results of the
162+
* operations. Note that the order of the results corresponds to the order of the operations.
163+
*
164+
* @param operations a list of {@code Operation} commands
165+
* @return a list of {@code BatchResult} that contains results of the operations
166+
* @throws E if any of the transaction CRUD operations fails
167+
*/
168+
List<BatchResult> batch(List<? extends Operation> operations) throws E;
169+
159170
/** A scanner abstraction for iterating results. */
160171
interface Scanner<E extends TransactionException> extends AutoCloseable, Iterable<Result> {
161172
/**
@@ -183,4 +194,41 @@ interface Scanner<E extends TransactionException> extends AutoCloseable, Iterabl
183194
@Override
184195
void close() throws E;
185196
}
197+
198+
/**
199+
* A batch operation result returned by {@link CrudOperable#batch(List)}. Represents the result of
200+
* a get operation or a scan operation.
201+
*/
202+
interface BatchResult {
203+
/**
204+
* Returns the type of the operation.
205+
*
206+
* @return the operation type
207+
*/
208+
Type getType();
209+
210+
/**
211+
* Returns a result of a get operation.
212+
*
213+
* @return an {@code Optional} with the returned result
214+
*/
215+
Optional<Result> getGetResult();
216+
217+
/**
218+
* Returns a list of results of a scan operation.
219+
*
220+
* @return a list of {@link Result}
221+
*/
222+
List<Result> getScanResult();
223+
224+
enum Type {
225+
GET,
226+
SCAN,
227+
PUT,
228+
INSERT,
229+
UPSERT,
230+
UPDATE,
231+
DELETE
232+
}
233+
}
186234
}

‎core/src/main/java/com/scalar/db/api/TransactionCrudOperable.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -167,6 +167,22 @@ void delete(List<Delete> deletes)
167167
void mutate(List<? extends Mutation> mutations)
168168
throws CrudConflictException, CrudException, UnsatisfiedConditionException;
169169

170+
/**
171+
* {@inheritDoc}
172+
*
173+
* @throws CrudConflictException if the transaction CRUD operation fails due to transient faults
174+
* (e.g., a conflict error). You can retry the transaction from the beginning
175+
* @throws CrudException if the transaction CRUD operation fails due to transient or nontransient
176+
* faults. You can try retrying the transaction from the beginning, but the transaction may
177+
* still fail if the cause is nontransient
178+
* @throws UnsatisfiedConditionException if a condition is specified in a {@link Put}, {@link
179+
* Delete}, or {@link Update} command, and if the condition is not satisfied or the entry does
180+
* not exist
181+
*/
182+
@Override
183+
List<BatchResult> batch(List<? extends Operation> operations)
184+
throws CrudConflictException, CrudException, UnsatisfiedConditionException;
185+
170186
interface Scanner extends CrudOperable.Scanner<CrudException> {
171187
/**
172188
* {@inheritDoc}

‎core/src/main/java/com/scalar/db/api/TransactionManagerCrudOperable.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -190,6 +190,24 @@ void mutate(List<? extends Mutation> mutations)
190190
throws CrudConflictException, CrudException, UnsatisfiedConditionException,
191191
UnknownTransactionStatusException;
192192

193+
/**
194+
* {@inheritDoc}
195+
*
196+
* @throws CrudConflictException if the transaction CRUD operation fails due to transient faults
197+
* (e.g., a conflict error). You can retry the transaction from the beginning
198+
* @throws CrudException if the transaction CRUD operation fails due to transient or nontransient
199+
* faults. You can try retrying the transaction from the beginning, but the transaction may
200+
* still fail if the cause is nontransient
201+
* @throws UnsatisfiedConditionException if a condition is specified in a {@link Put}, {@link
202+
* Delete}, or {@link Update} command, and if the condition is not satisfied or the entry does
203+
* not exist
204+
* @throws UnknownTransactionStatusException if the status of the commit is unknown
205+
*/
206+
@Override
207+
List<BatchResult> batch(List<? extends Operation> operations)
208+
throws CrudConflictException, CrudException, UnsatisfiedConditionException,
209+
UnknownTransactionStatusException;
210+
193211
interface Scanner extends CrudOperable.Scanner<TransactionException> {
194212
/**
195213
* {@inheritDoc}

‎core/src/main/java/com/scalar/db/common/AbstractDistributedTransaction.java‎

Lines changed: 60 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,21 @@
11
package com.scalar.db.common;
22

3+
import static com.google.common.base.Preconditions.checkArgument;
4+
35
import com.scalar.db.api.Delete;
46
import com.scalar.db.api.DistributedTransaction;
57
import com.scalar.db.api.Get;
68
import com.scalar.db.api.Insert;
79
import com.scalar.db.api.Mutation;
10+
import com.scalar.db.api.Operation;
811
import com.scalar.db.api.Put;
12+
import com.scalar.db.api.Result;
913
import com.scalar.db.api.Scan;
1014
import com.scalar.db.api.Update;
1115
import com.scalar.db.api.Upsert;
16+
import com.scalar.db.exception.transaction.CrudException;
1217
import com.scalar.db.util.ScalarDbUtils;
18+
import java.util.ArrayList;
1319
import java.util.List;
1420
import java.util.Optional;
1521

@@ -59,8 +65,60 @@ public Optional<String> getTable() {
5965
return tableName;
6066
}
6167

62-
protected <T extends Mutation> List<T> copyAndSetTargetToIfNot(List<T> mutations) {
63-
return ScalarDbUtils.copyAndSetTargetToIfNot(mutations, namespace, tableName);
68+
@Override
69+
public void mutate(List<? extends Mutation> mutations) throws CrudException {
70+
checkArgument(!mutations.isEmpty(), CoreError.EMPTY_MUTATIONS_SPECIFIED.buildMessage());
71+
for (Mutation mutation : mutations) {
72+
if (mutation instanceof Put) {
73+
put((Put) mutation);
74+
} else if (mutation instanceof Delete) {
75+
delete((Delete) mutation);
76+
} else if (mutation instanceof Insert) {
77+
insert((Insert) mutation);
78+
} else if (mutation instanceof Upsert) {
79+
upsert((Upsert) mutation);
80+
} else {
81+
assert mutation instanceof Update;
82+
update((Update) mutation);
83+
}
84+
}
85+
}
86+
87+
@Override
88+
public List<BatchResult> batch(List<? extends Operation> operations) throws CrudException {
89+
checkArgument(!operations.isEmpty(), CoreError.EMPTY_OPERATIONS_SPECIFIED.buildMessage());
90+
List<BatchResult> ret = new ArrayList<>();
91+
for (Operation operation : operations) {
92+
if (operation instanceof Get) {
93+
Optional<Result> result = get((Get) operation);
94+
ret.add(new BatchResultImpl(result));
95+
} else if (operation instanceof Scan) {
96+
List<Result> results = scan((Scan) operation);
97+
ret.add(new BatchResultImpl(results));
98+
} else if (operation instanceof Put) {
99+
put((Put) operation);
100+
ret.add(BatchResultImpl.PUT_BATCH_RESULT);
101+
} else if (operation instanceof Insert) {
102+
insert((Insert) operation);
103+
ret.add(BatchResultImpl.INSERT_BATCH_RESULT);
104+
} else if (operation instanceof Upsert) {
105+
upsert((Upsert) operation);
106+
ret.add(BatchResultImpl.UPSERT_BATCH_RESULT);
107+
} else if (operation instanceof Update) {
108+
update((Update) operation);
109+
ret.add(BatchResultImpl.UPDATE_BATCH_RESULT);
110+
} else if (operation instanceof Delete) {
111+
delete((Delete) operation);
112+
ret.add(BatchResultImpl.DELETE_BATCH_RESULT);
113+
} else {
114+
throw new AssertionError("Unknown operation: " + operation);
115+
}
116+
}
117+
return ret;
118+
}
119+
120+
protected <T extends Operation> List<T> copyAndSetTargetToIfNot(List<T> operations) {
121+
return ScalarDbUtils.copyAndSetTargetToIfNot(operations, namespace, tableName);
64122
}
65123

66124
protected Get copyAndSetTargetToIfNot(Get get) {

‎core/src/main/java/com/scalar/db/common/AbstractDistributedTransactionManager.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
import com.scalar.db.api.DistributedTransactionManager;
66
import com.scalar.db.api.Get;
77
import com.scalar.db.api.Insert;
8-
import com.scalar.db.api.Mutation;
8+
import com.scalar.db.api.Operation;
99
import com.scalar.db.api.Put;
1010
import com.scalar.db.api.Scan;
1111
import com.scalar.db.api.Update;
@@ -73,8 +73,8 @@ public DistributedTransaction resume(String txId) throws TransactionNotFoundExce
7373
throw new UnsupportedOperationException("resume is not supported in this implementation");
7474
}
7575

76-
protected <T extends Mutation> List<T> copyAndSetTargetToIfNot(List<T> mutations) {
77-
return ScalarDbUtils.copyAndSetTargetToIfNot(mutations, namespace, tableName);
76+
protected <T extends Operation> List<T> copyAndSetTargetToIfNot(List<T> operations) {
77+
return ScalarDbUtils.copyAndSetTargetToIfNot(operations, namespace, tableName);
7878
}
7979

8080
protected Get copyAndSetTargetToIfNot(Get get) {

‎core/src/main/java/com/scalar/db/common/AbstractTwoPhaseCommitTransaction.java‎

Lines changed: 60 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,21 @@
11
package com.scalar.db.common;
22

3+
import static com.google.common.base.Preconditions.checkArgument;
4+
35
import com.scalar.db.api.Delete;
46
import com.scalar.db.api.Get;
57
import com.scalar.db.api.Insert;
68
import com.scalar.db.api.Mutation;
9+
import com.scalar.db.api.Operation;
710
import com.scalar.db.api.Put;
11+
import com.scalar.db.api.Result;
812
import com.scalar.db.api.Scan;
913
import com.scalar.db.api.TwoPhaseCommitTransaction;
1014
import com.scalar.db.api.Update;
1115
import com.scalar.db.api.Upsert;
16+
import com.scalar.db.exception.transaction.CrudException;
1217
import com.scalar.db.util.ScalarDbUtils;
18+
import java.util.ArrayList;
1319
import java.util.List;
1420
import java.util.Optional;
1521

@@ -59,8 +65,60 @@ public Optional<String> getTable() {
5965
return tableName;
6066
}
6167

62-
protected <T extends Mutation> List<T> copyAndSetTargetToIfNot(List<T> mutations) {
63-
return ScalarDbUtils.copyAndSetTargetToIfNot(mutations, namespace, tableName);
68+
@Override
69+
public void mutate(List<? extends Mutation> mutations) throws CrudException {
70+
checkArgument(!mutations.isEmpty(), CoreError.EMPTY_MUTATIONS_SPECIFIED.buildMessage());
71+
for (Mutation mutation : mutations) {
72+
if (mutation instanceof Put) {
73+
put((Put) mutation);
74+
} else if (mutation instanceof Delete) {
75+
delete((Delete) mutation);
76+
} else if (mutation instanceof Insert) {
77+
insert((Insert) mutation);
78+
} else if (mutation instanceof Upsert) {
79+
upsert((Upsert) mutation);
80+
} else {
81+
assert mutation instanceof Update;
82+
update((Update) mutation);
83+
}
84+
}
85+
}
86+
87+
@Override
88+
public List<BatchResult> batch(List<? extends Operation> operations) throws CrudException {
89+
checkArgument(!operations.isEmpty(), CoreError.EMPTY_OPERATIONS_SPECIFIED.buildMessage());
90+
List<BatchResult> ret = new ArrayList<>();
91+
for (Operation operation : operations) {
92+
if (operation instanceof Get) {
93+
Optional<Result> result = get((Get) operation);
94+
ret.add(new BatchResultImpl(result));
95+
} else if (operation instanceof Scan) {
96+
List<Result> results = scan((Scan) operation);
97+
ret.add(new BatchResultImpl(results));
98+
} else if (operation instanceof Put) {
99+
put((Put) operation);
100+
ret.add(BatchResultImpl.PUT_BATCH_RESULT);
101+
} else if (operation instanceof Insert) {
102+
insert((Insert) operation);
103+
ret.add(BatchResultImpl.INSERT_BATCH_RESULT);
104+
} else if (operation instanceof Upsert) {
105+
upsert((Upsert) operation);
106+
ret.add(BatchResultImpl.UPSERT_BATCH_RESULT);
107+
} else if (operation instanceof Update) {
108+
update((Update) operation);
109+
ret.add(BatchResultImpl.UPDATE_BATCH_RESULT);
110+
} else if (operation instanceof Delete) {
111+
delete((Delete) operation);
112+
ret.add(BatchResultImpl.DELETE_BATCH_RESULT);
113+
} else {
114+
throw new AssertionError("Unknown operation: " + operation);
115+
}
116+
}
117+
return ret;
118+
}
119+
120+
protected <T extends Operation> List<T> copyAndSetTargetToIfNot(List<T> operations) {
121+
return ScalarDbUtils.copyAndSetTargetToIfNot(operations, namespace, tableName);
64122
}
65123

66124
protected Get copyAndSetTargetToIfNot(Get get) {

‎core/src/main/java/com/scalar/db/common/AbstractTwoPhaseCommitTransactionManager.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
import com.scalar.db.api.Delete;
44
import com.scalar.db.api.Get;
55
import com.scalar.db.api.Insert;
6-
import com.scalar.db.api.Mutation;
6+
import com.scalar.db.api.Operation;
77
import com.scalar.db.api.Put;
88
import com.scalar.db.api.Scan;
99
import com.scalar.db.api.TwoPhaseCommitTransaction;
@@ -68,8 +68,8 @@ public TwoPhaseCommitTransaction resume(String txId) throws TransactionNotFoundE
6868
throw new UnsupportedOperationException("resume is not supported in this implementation");
6969
}
7070

71-
protected <T extends Mutation> List<T> copyAndSetTargetToIfNot(List<T> mutations) {
72-
return ScalarDbUtils.copyAndSetTargetToIfNot(mutations, namespace, tableName);
71+
protected <T extends Operation> List<T> copyAndSetTargetToIfNot(List<T> operations) {
72+
return ScalarDbUtils.copyAndSetTargetToIfNot(operations, namespace, tableName);
7373
}
7474

7575
protected Get copyAndSetTargetToIfNot(Get get) {

0 commit comments

Comments
 (0)